feat: sync DingTalk QA to Notable question table

This commit is contained in:
姚章浩
2026-07-23 14:38:50 +08:00
parent 5e0694835a
commit 0a3f62618e
22 changed files with 4351 additions and 77 deletions
+12
View File
@@ -28,6 +28,18 @@ Implemented:
- Required env vars to enable (all three or none): `DINGTALK_CLIENT_ID`, `DINGTALK_CLIENT_SECRET`, `DOBOTAI_TENANT_KEY`. Partial configuration is a fatal startup error.
- Optional `DINGTALK_SUPPORT_GROUP_OPEN_CONVERSATION_ID` enables durable uncovered-answer notifications through the official group-message API. When empty, coverage decisions and pending support cases are still recorded but no group message is sent.
## DingTalk question table sync (Notable v1)
Successful DingTalk QA turns can be appended asynchronously to a DingTalk Notable AI table through a durable PostgreSQL outbox. Notable failures never change the successful QA result; the worker retries with bounded backoff and stores only stable failure codes.
Apply `db/migrations/000010_question_table_exports.up.sql` before enabling the worker. The three table settings are optional as a group: all blank disables dispatch, while a partial configuration fails startup closed.
- `DINGTALK_QUESTION_TABLE_BASE_ID`
- `DINGTALK_QUESTION_TABLE_SHEET_ID`
- `DINGTALK_QUESTION_TABLE_OPERATOR_ID` (DingTalk unionId)
The worker reuses `DINGTALK_CLIENT_ID` and `DINGTALK_CLIENT_SECRET`. The DingTalk application must have `Notable.Base.Read.All` and `Notable.Base.Write.All`. Only successful `channel=dingtalk` turns are enqueued; other channels are excluded. The first release writes the nine basic text/date/user fields and omits the five select-classification fields.
## WeKnora baseline
Target policy: latest stable official Release/Tag. The newest locally fetched stable tag is `v0.6.3`. Online GitHub verification is pending because DNS/proxy access was unavailable during initialization.
+29
View File
@@ -99,6 +99,28 @@ func main() {
go runSupportDispatcher(ctx, supportDispatcher, 5*time.Second)
log.Printf("DingTalk support notification dispatcher started")
}
// Question table dispatcher (optional, fail-closed).
qtCfg, err := readQuestionTableConfig()
if err != nil {
log.Fatal(err)
}
var questionTableDone chan struct{}
if qtCfg.BaseID != "" {
qtRepo, err := postgres.NewQuestionTableExportRepository(store)
if err != nil {
log.Fatal(err)
}
qtDispatcher, err := buildQuestionTableDispatcher(qtRepo, dingCfg.ClientID, dingCfg.ClientSecret, qtCfg)
if err != nil {
log.Fatalf("DingTalk question table dispatcher init failed: %v", err)
}
questionTableDone = make(chan struct{})
go func() {
defer close(questionTableDone)
runQuestionTablePoller(ctx, qtDispatcher, 5*time.Second)
}()
log.Printf("DingTalk question table dispatcher started")
}
var streamClient *dingtalk.StreamClient
if dingCfg.Enabled() {
directoryLookup, err := directory.NewDingTalkLookup(directory.DingTalkLookupConfig{
@@ -182,6 +204,13 @@ func main() {
if err := server.Shutdown(shutdownCtx); err != nil {
log.Printf("HTTP shutdown: %v", err)
}
if questionTableDone != nil {
select {
case <-questionTableDone:
case <-shutdownCtx.Done():
log.Printf("question table poller shutdown timed out")
}
}
}
func buildAuditOptions(repo audit.TurnRepository, tenantKey, adminToken, address string) (httpapi.Options, error) {
+142
View File
@@ -0,0 +1,142 @@
package main
import (
"context"
"errors"
"log"
"os"
"reflect"
"strings"
"time"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/questiontable"
)
// questionTablePollerRunner is the interface expected by the poller loop.
type questionTablePollerRunner interface {
RunOnce(context.Context) (questiontable.DispatcherStats, error)
}
// questionTableConfig holds the optional Notable question-table env values.
type questionTableConfig struct {
BaseID string
SheetID string
OperatorID string
}
// readQuestionTableConfig reads the three dedicated env vars and returns
// a validated config. All three blank => disabled (no error). Any partial
// configuration => stable configuration error (no values echoed).
func readQuestionTableConfig() (questionTableConfig, error) {
cfg := questionTableConfig{
BaseID: strings.TrimSpace(os.Getenv("DINGTALK_QUESTION_TABLE_BASE_ID")),
SheetID: strings.TrimSpace(os.Getenv("DINGTALK_QUESTION_TABLE_SHEET_ID")),
OperatorID: strings.TrimSpace(os.Getenv("DINGTALK_QUESTION_TABLE_OPERATOR_ID")),
}
allBlank := cfg.BaseID == "" && cfg.SheetID == "" && cfg.OperatorID == ""
allSet := cfg.BaseID != "" && cfg.SheetID != "" && cfg.OperatorID != ""
if allBlank {
return questionTableConfig{}, nil
}
if allSet {
return cfg, nil
}
return questionTableConfig{}, errors.New("DingTalk question table config: all three env vars must be set or all blank (DINGTALK_QUESTION_TABLE_BASE_ID, DINGTALK_QUESTION_TABLE_SHEET_ID, DINGTALK_QUESTION_TABLE_OPERATOR_ID)")
}
// buildQuestionTableDispatcher creates a validated question-table dispatcher
// when the repo is non-nil and all required credentials are present.
// No HTTP request occurs at build time. It returns (nil, nil) when the
// question-table feature is disabled (empty config).
func buildQuestionTableDispatcher(repo questiontable.ExportRepository, clientID, clientSecret string, cfg questionTableConfig) (*questiontable.Dispatcher, error) {
// Disabled: all blank config.
if cfg.BaseID == "" && cfg.SheetID == "" && cfg.OperatorID == "" {
return nil, nil
}
if repo == nil {
return nil, errors.New("question table dispatcher: repository is required")
}
client, err := questiontable.NewClient(questiontable.Config{
ClientID: clientID,
Secret: clientSecret,
BaseID: cfg.BaseID,
SheetID: cfg.SheetID,
OperatorID: cfg.OperatorID,
}, nil)
if err != nil {
return nil, err
}
return questiontable.NewDispatcher(repo, client, nil)
}
// runQuestionTablePoller runs the dispatcher immediately, then every interval.
// Non-positive interval or nil/typed-nil runner returns without calling.
// Continues after a RunOnce error. Exits promptly on context cancellation.
// Logging is safe: success may log only Claimed/Exported/Failed integers;
// errors log stable stage text only, never raw errors.
func runQuestionTablePoller(ctx context.Context, runner questionTablePollerRunner, interval time.Duration) {
if runner == nil || interval <= 0 {
return
}
// Guard against typed nil.
if isNilPoller(runner) {
return
}
runOnce := func() {
runCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
stats, err := runner.RunOnce(runCtx)
cancel()
if err != nil {
log.Printf("question table dispatch failed")
return
}
if stats.Claimed > 0 || stats.Exported > 0 || stats.Failed > 0 {
log.Printf("question table dispatch: claimed=%d exported=%d failed=%d", stats.Claimed, stats.Exported, stats.Failed)
}
}
if ctx.Err() != nil {
return
}
runOnce()
ticker := time.NewTicker(interval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
runOnce()
}
}
}
// isNilPoller returns true when runner is a typed nil.
func isNilPoller(runner questionTablePollerRunner) bool {
if runner == nil {
return true
}
// reflect-based typed-nil check for the interface.
return isInterfaceNil(runner)
}
// isInterfaceNil returns true when v is a typed nil wrapped in an interface.
func isInterfaceNil(v interface{}) bool {
if v == nil {
return true
}
rv := reflect.ValueOf(v)
switch rv.Kind() {
case reflect.Chan, reflect.Func, reflect.Interface, reflect.Map, reflect.Ptr, reflect.Slice:
return rv.IsNil()
default:
return false
}
}
+377
View File
@@ -0,0 +1,377 @@
package main
import (
"context"
"errors"
"log"
"os"
"strings"
"sync"
"testing"
"time"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/questiontable"
)
// ---------------------------------------------------------------------------
// Test 1: disabled all blank returns nil/no error
// ---------------------------------------------------------------------------
func TestReadQuestionTableConfigDisabledAllBlank(t *testing.T) {
t.Setenv("DINGTALK_QUESTION_TABLE_BASE_ID", "")
t.Setenv("DINGTALK_QUESTION_TABLE_SHEET_ID", "")
t.Setenv("DINGTALK_QUESTION_TABLE_OPERATOR_ID", "")
cfg, err := readQuestionTableConfig()
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if cfg.BaseID != "" || cfg.SheetID != "" || cfg.OperatorID != "" {
t.Errorf("expected empty config, got %+v", cfg)
}
}
func TestBuildQuestionTableDispatcherDisabled(t *testing.T) {
d, err := buildQuestionTableDispatcher(nil, "", "", questionTableConfig{})
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if d != nil {
t.Errorf("expected nil dispatcher for disabled config, got %v", d)
}
}
// ---------------------------------------------------------------------------
// Test 2: every partial combination errors and does not echo sentinel config values
// ---------------------------------------------------------------------------
func TestReadQuestionTableConfigPartialErrors(t *testing.T) {
tests := []struct {
name string
baseID string
sheetID string
operator string
}{
{"only base", "base-1", "", ""},
{"only sheet", "", "sheet-1", ""},
{"only operator", "", "", "op-1"},
{"base+sheet", "base-1", "sheet-1", ""},
{"base+operator", "base-1", "", "op-1"},
{"sheet+operator", "", "sheet-1", "op-1"},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
t.Setenv("DINGTALK_QUESTION_TABLE_BASE_ID", tc.baseID)
t.Setenv("DINGTALK_QUESTION_TABLE_SHEET_ID", tc.sheetID)
t.Setenv("DINGTALK_QUESTION_TABLE_OPERATOR_ID", tc.operator)
_, err := readQuestionTableConfig()
if err == nil {
t.Fatal("expected error for partial config")
}
errStr := err.Error()
// Must not echo any sentinel config value.
for _, sentinel := range []string{"base-1", "sheet-1", "op-1"} {
if strings.Contains(errStr, sentinel) {
t.Errorf("error must not echo sentinel value %q: %q", sentinel, errStr)
}
}
})
}
}
// ---------------------------------------------------------------------------
// Test 3: enabled missing repo/client/secret fails closed; complete config builds
// ---------------------------------------------------------------------------
func TestBuildQuestionTableDispatcherEnabledMissingRepo(t *testing.T) {
cfg := questionTableConfig{BaseID: "b1", SheetID: "sh1", OperatorID: "op1"}
_, err := buildQuestionTableDispatcher(nil, "cid", "secret", cfg)
if err == nil {
t.Fatal("expected error when repo is nil")
}
if !strings.Contains(err.Error(), "repository is required") {
t.Errorf("error = %v, want 'repository is required'", err)
}
}
func TestBuildQuestionTableDispatcherEnabledMissingClientID(t *testing.T) {
repo := &questionTableExportRepoStub{}
cfg := questionTableConfig{BaseID: "b1", SheetID: "sh1", OperatorID: "op1"}
_, err := buildQuestionTableDispatcher(repo, "", "secret", cfg)
if err == nil {
t.Fatal("expected error when client id is empty")
}
}
func TestBuildQuestionTableDispatcherEnabledMissingSecret(t *testing.T) {
repo := &questionTableExportRepoStub{}
cfg := questionTableConfig{BaseID: "b1", SheetID: "sh1", OperatorID: "op1"}
_, err := buildQuestionTableDispatcher(repo, "cid", "", cfg)
if err == nil {
t.Fatal("expected error when secret is empty")
}
}
func TestBuildQuestionTableDispatcherEnabledCompleteConfig(t *testing.T) {
repo := &questionTableExportRepoStub{}
cfg := questionTableConfig{BaseID: "b1", SheetID: "sh1", OperatorID: "op1"}
d, err := buildQuestionTableDispatcher(repo, "cid", "secret", cfg)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if d == nil {
t.Fatal("expected non-nil dispatcher for complete config")
}
}
// ---------------------------------------------------------------------------
// Test 4: immediate run + periodic repeat + cancel stop
// ---------------------------------------------------------------------------
func TestRunQuestionTablePollerImmediateAndPeriodic(t *testing.T) {
runner := &countingPollerRunner{}
ctx, cancel := context.WithCancel(context.Background())
var wg sync.WaitGroup
wg.Add(1)
go func() {
defer wg.Done()
runQuestionTablePoller(ctx, runner, 50*time.Millisecond)
}()
// Wait for at least 2 calls (immediate + one tick).
time.Sleep(120 * time.Millisecond)
cancel()
wg.Wait()
runner.mu.Lock()
calls := runner.calls
runner.mu.Unlock()
if calls < 2 {
t.Errorf("expected at least 2 calls, got %d", calls)
}
}
func TestRunQuestionTablePollerCancelledContext(t *testing.T) {
runner := &countingPollerRunner{}
ctx, cancel := context.WithCancel(context.Background())
cancel() // already cancelled
runQuestionTablePoller(ctx, runner, 50*time.Millisecond)
runner.mu.Lock()
calls := runner.calls
runner.mu.Unlock()
if calls != 0 {
t.Errorf("expected 0 calls (cancelled context), got %d", calls)
}
}
// ---------------------------------------------------------------------------
// Test 5: continue after error and log/output security contract
// ---------------------------------------------------------------------------
func TestRunQuestionTablePollerContinuesAfterError(t *testing.T) {
runner := &errorThenSuccessPoller{}
ctx, cancel := context.WithCancel(context.Background())
var wg sync.WaitGroup
wg.Add(1)
go func() {
defer wg.Done()
runQuestionTablePoller(ctx, runner, 30*time.Millisecond)
}()
time.Sleep(120 * time.Millisecond)
cancel()
wg.Wait()
runner.mu.Lock()
calls := runner.calls
runner.mu.Unlock()
if calls < 2 {
t.Errorf("expected at least 2 calls after error, got %d", calls)
}
}
// Test logging contract: the runOnce function in runQuestionTablePoller logs
// only stable stage text for errors (never raw errors via %v) and only
// Claimed/Exported/Failed integers for success.
func TestPollerLoggingContract(t *testing.T) {
// Behavioral log-capture test: runner returns an error containing unique
// sentinel values; the captured log must contain the stable stage text
// and none of those sentinels.
sentinel := "s3nt1n3l-op3r4t0r-t0k3n-q"
runner := &sentinelErrorPoller{errMsg: "questiontable dispatcher: " + sentinel}
ctx, cancel := context.WithCancel(context.Background())
// Save and restore global log output.
originalOutput := log.Writer()
originalFlags := log.Flags()
var buf strings.Builder
log.SetOutput(&buf)
log.SetFlags(0)
defer func() {
log.SetOutput(originalOutput)
log.SetFlags(originalFlags)
}()
var wg sync.WaitGroup
wg.Add(1)
go func() {
defer wg.Done()
runQuestionTablePoller(ctx, runner, 50*time.Millisecond)
}()
time.Sleep(120 * time.Millisecond)
cancel()
wg.Wait()
output := buf.String()
if !strings.Contains(output, "question table dispatch failed") {
t.Errorf("log must contain stable stage text, got: %s", output)
}
if strings.Contains(output, sentinel) {
t.Errorf("log must not contain sentinel %q, got: %s", sentinel, output)
}
}
// sentinelErrorPoller always returns an error with the given message.
type sentinelErrorPoller struct {
errMsg string
}
func (r *sentinelErrorPoller) RunOnce(ctx context.Context) (questiontable.DispatcherStats, error) {
return questiontable.DispatcherStats{}, errors.New(r.errMsg)
}
// ---------------------------------------------------------------------------
// Test 6: nil/typed-nil runner and nonpositive interval safe
// ---------------------------------------------------------------------------
func TestRunQuestionTablePollerNilRunner(t *testing.T) {
ctx := context.Background()
// Should not panic, just return immediately.
runQuestionTablePoller(ctx, nil, 50*time.Millisecond)
// If we reach here without panic, test passes.
}
func TestRunQuestionTablePollerNonPositiveInterval(t *testing.T) {
runner := &countingPollerRunner{}
ctx := context.Background()
runQuestionTablePoller(ctx, runner, 0)
runQuestionTablePoller(ctx, runner, -1*time.Second)
runner.mu.Lock()
calls := runner.calls
runner.mu.Unlock()
if calls != 0 {
t.Errorf("expected 0 calls for non-positive interval, got %d", calls)
}
}
func TestRunQuestionTablePollerTypedNilInterface(t *testing.T) {
// The runner is nil through the interface but not at the concrete type level.
var typedNil *questiontable.Dispatcher = nil
var runner questionTablePollerRunner = typedNil
ctx := context.Background()
// Should not panic, just return immediately.
runQuestionTablePoller(ctx, runner, 50*time.Millisecond)
}
// ---------------------------------------------------------------------------
// Test 7: main source wiring contract
// ---------------------------------------------------------------------------
func TestMainWiresQuestionTableEnvNames(t *testing.T) {
mainData, err := os.ReadFile("main.go")
if err != nil {
t.Fatal(err)
}
syncData, err := os.ReadFile("question_table_sync.go")
if err != nil {
t.Fatal(err)
}
mainSource := string(mainData)
syncSource := string(syncData)
for _, fragment := range []string{
`"DINGTALK_QUESTION_TABLE_BASE_ID"`,
`"DINGTALK_QUESTION_TABLE_SHEET_ID"`,
`"DINGTALK_QUESTION_TABLE_OPERATOR_ID"`,
} {
if !strings.Contains(syncSource, fragment) {
t.Errorf("question table config reader missing %q", fragment)
}
}
for _, fragment := range []string{
`NewQuestionTableExportRepository`,
`5*time.Second`,
`"DingTalk question table dispatcher started"`,
} {
if !strings.Contains(mainSource, fragment) {
t.Errorf("main wiring missing %q", fragment)
}
}
shutdownAt := strings.Index(mainSource, "server.Shutdown(shutdownCtx)")
waitAt := strings.Index(mainSource, "case <-questionTableDone:")
if shutdownAt < 0 || waitAt < 0 || waitAt <= shutdownAt {
t.Fatal("main must wait for question table poller after HTTP shutdown and before returning")
}
}
// ---------------------------------------------------------------------------
// Test doubles
// ---------------------------------------------------------------------------
// questionTableExportRepoStub implements questiontable.ExportRepository.
type questionTableExportRepoStub struct{}
func (r *questionTableExportRepoStub) Claim(ctx context.Context, limit int, maxAttempts int, staleBefore time.Time) ([]questiontable.PendingExport, error) {
return nil, nil
}
func (r *questionTableExportRepoStub) MarkExported(ctx context.Context, exportID, location string) error {
return nil
}
func (r *questionTableExportRepoStub) MarkFailed(ctx context.Context, exportID, failureCode string) error {
return nil
}
// countingPollerRunner counts calls to RunOnce.
type countingPollerRunner struct {
mu sync.Mutex
calls int
}
func (r *countingPollerRunner) RunOnce(ctx context.Context) (questiontable.DispatcherStats, error) {
r.mu.Lock()
r.calls++
r.mu.Unlock()
return questiontable.DispatcherStats{}, nil
}
// errorThenSuccessPoller returns an error on the first call, then succeeds.
type errorThenSuccessPoller struct {
mu sync.Mutex
calls int
}
func (r *errorThenSuccessPoller) RunOnce(ctx context.Context) (questiontable.DispatcherStats, error) {
r.mu.Lock()
r.calls++
call := r.calls
r.mu.Unlock()
if call == 1 {
return questiontable.DispatcherStats{}, errors.New("questiontable dispatcher: claim failed")
}
return questiontable.DispatcherStats{Claimed: 1, Exported: 1}, nil
}
@@ -0,0 +1,9 @@
-- 000010_question_table_exports down migration
-- Drop in reverse dependency order: first drop export table, then remove column
DROP TABLE IF EXISTS question_table_exports;
ALTER TABLE messages
DROP COLUMN IF EXISTS trigger_person_snapshot;
ALTER TABLE messages
DROP COLUMN IF EXISTS actor_union_id_snapshot;
@@ -0,0 +1,40 @@
-- 000010_question_table_exports: durable outbox for Notable question table async sync
-- Add trigger_person_snapshot and actor_union_id_snapshot to messages and create question_table_exports table
-- 1. Add trigger_person_snapshot and actor_union_id_snapshot columns to messages
ALTER TABLE messages
ADD COLUMN IF NOT EXISTS trigger_person_snapshot TEXT NOT NULL DEFAULT '';
ALTER TABLE messages
ADD COLUMN IF NOT EXISTS actor_union_id_snapshot TEXT NOT NULL DEFAULT '';
-- 2. Create question_table_exports outbox table
CREATE TABLE IF NOT EXISTS question_table_exports (
id TEXT PRIMARY KEY,
tenant_key TEXT NOT NULL,
channel TEXT NOT NULL,
turn_id TEXT NOT NULL REFERENCES turns(id),
status TEXT NOT NULL DEFAULT 'pending'
CHECK (status IN ('pending', 'exporting', 'exported', 'export_failed')),
attempts INTEGER NOT NULL DEFAULT 0,
last_attempt_at TIMESTAMPTZ,
next_attempt_at TIMESTAMPTZ,
exported_at TIMESTAMPTZ,
failure_code TEXT NOT NULL DEFAULT '',
external_record_id TEXT NOT NULL DEFAULT '',
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
UNIQUE (tenant_key, channel, turn_id)
);
-- 3. Index for claim query: pending/export_failed/exporting ordered by created_at
-- Covers pending, export_failed (when next_attempt_at is NULL or <= NOW()),
-- and stale exporting (when last_attempt_at < staleBefore).
CREATE INDEX IF NOT EXISTS idx_question_table_exports_claim
ON question_table_exports (status, next_attempt_at, created_at, id)
WHERE status IN ('pending', 'export_failed')
OR (status = 'exporting' AND last_attempt_at IS NOT NULL);
-- 4. Index for stale exporting recovery
CREATE INDEX IF NOT EXISTS idx_question_table_exports_stale
ON question_table_exports (status, last_attempt_at)
WHERE status = 'exporting';
@@ -0,0 +1,161 @@
// Package migrations_test contains source-level contract tests for the
// 000010_question_table_exports migration. No real database connection is used.
package migrations_test
import (
"os"
"path/filepath"
"strings"
"testing"
)
func read010Migration(t *testing.T, direction string) string {
t.Helper()
data, err := os.ReadFile(filepath.Join(".", "000010_question_table_exports."+direction+".sql"))
if err != nil {
t.Fatalf("read 000010 %s migration: %v", direction, err)
}
return string(data)
}
func TestUp010AddsTriggerPersonSnapshot(t *testing.T) {
up := strings.ToUpper(read010Migration(t, "up"))
if !strings.Contains(up, "ALTER TABLE MESSAGES") {
t.Fatal("up migration must alter messages")
}
if !strings.Contains(up, "ADD COLUMN IF NOT EXISTS TRIGGER_PERSON_SNAPSHOT") {
t.Error("up migration must add trigger_person_snapshot column")
}
if !strings.Contains(up, "ADD COLUMN IF NOT EXISTS ACTOR_UNION_ID_SNAPSHOT") {
t.Error("up migration must add actor_union_id_snapshot column")
}
if !strings.Contains(up, "TEXT NOT NULL DEFAULT ''") {
t.Error("snapshot columns must be TEXT NOT NULL DEFAULT ''")
}
}
func TestUp010CreatesQuestionTableExports(t *testing.T) {
up := strings.ToUpper(read010Migration(t, "up"))
required := []string{
"CREATE TABLE IF NOT EXISTS QUESTION_TABLE_EXPORTS",
"ID", "TENANT_KEY", "CHANNEL", "TURN_ID",
"STATUS", "ATTEMPTS", "LAST_ATTEMPT_AT", "NEXT_ATTEMPT_AT", "EXPORTED_AT",
"FAILURE_CODE", "EXTERNAL_RECORD_ID",
"CREATED_AT", "UPDATED_AT",
}
for _, fragment := range required {
if !strings.Contains(up, fragment) {
t.Errorf("up migration missing %q", fragment)
}
}
}
func TestUp010QuestionTableExportsConstraints(t *testing.T) {
up := strings.ToUpper(read010Migration(t, "up"))
// FK to turns
if !strings.Contains(up, "REFERENCES TURNS(ID)") {
t.Error("question_table_exports.turn_id must reference turns(id)")
}
// UNIQUE (tenant_key, channel, turn_id)
if !strings.Contains(up, "UNIQUE (TENANT_KEY, CHANNEL, TURN_ID)") {
t.Error("question_table_exports must have UNIQUE (tenant_key, channel, turn_id)")
}
// status CHECK
for _, status := range []string{"PENDING", "EXPORTING", "EXPORTED", "EXPORT_FAILED"} {
if !strings.Contains(up, status) {
t.Errorf("status CHECK must include %q", status)
}
}
}
func TestUp010ClaimIndexes(t *testing.T) {
up := strings.ToUpper(read010Migration(t, "up"))
if !strings.Contains(up, "IDX_QUESTION_TABLE_EXPORTS_CLAIM") {
t.Error("up migration must create claim index")
}
if !strings.Contains(up, "WHERE STATUS IN ('PENDING', 'EXPORT_FAILED')") {
t.Error("claim index must include pending and export_failed statuses")
}
if !strings.Contains(up, "IDX_QUESTION_TABLE_EXPORTS_STALE") {
t.Error("up migration must create stale exporting index")
}
if !strings.Contains(up, "WHERE STATUS = 'EXPORTING'") {
t.Error("stale index must filter exporting status")
}
}
func TestUp010HasNextAttemptAtColumn(t *testing.T) {
up := strings.ToUpper(read010Migration(t, "up"))
if !strings.Contains(up, "NEXT_ATTEMPT_AT TIMESTAMPTZ") {
t.Error("question_table_exports must have next_attempt_at column for retry backoff")
}
}
func TestUp010HasExternalRecordID(t *testing.T) {
up := strings.ToUpper(read010Migration(t, "up"))
if !strings.Contains(up, "EXTERNAL_RECORD_ID TEXT NOT NULL DEFAULT ''") {
t.Error("question_table_exports must have external_record_id column (not a1_notation)")
}
if strings.Contains(up, "A1_NOTATION") {
t.Error("question_table_exports must not have stale a1_notation column")
}
}
func TestUp010ClaimIndexCoversNextAttemptAt(t *testing.T) {
up := strings.ToUpper(read010Migration(t, "up"))
if !strings.Contains(up, "NEXT_ATTEMPT_AT") {
t.Error("claim index must cover next_attempt_at for retry backoff filtering")
}
}
func TestDown010DropsInSafeOrder(t *testing.T) {
down := strings.ToUpper(read010Migration(t, "down"))
dropExportAt := strings.Index(down, "DROP TABLE IF EXISTS QUESTION_TABLE_EXPORTS")
dropTriggerAt := strings.Index(down, "DROP COLUMN IF EXISTS TRIGGER_PERSON_SNAPSHOT")
dropUnionAt := strings.Index(down, "DROP COLUMN IF EXISTS ACTOR_UNION_ID_SNAPSHOT")
if dropExportAt < 0 || dropTriggerAt < 0 || dropUnionAt < 0 {
t.Fatal("down migration must drop export table and both snapshot columns")
}
if dropExportAt > dropTriggerAt || dropTriggerAt > dropUnionAt {
t.Fatal("down migration must drop export table before snapshot columns")
}
}
func TestDown010DoesNotDropBaselineTables(t *testing.T) {
down := strings.ToUpper(read010Migration(t, "down"))
for _, table := range []string{"TURNS", "MESSAGES", "SESSION_MAPPINGS", "CHANNEL_ACTORS", "CHANNEL_CONVERSATIONS"} {
if strings.Contains(down, "DROP TABLE IF EXISTS "+table) {
t.Errorf("down migration must not drop baseline table %q", table)
}
}
}
func TestUp010ExportsNoFormulaColumns(t *testing.T) {
up := read010Migration(t, "up")
// Check that the CREATE TABLE body does not contain user/AI content columns
lines := strings.Split(up, "\n")
inTable := false
for _, line := range lines {
trimmed := strings.TrimSpace(line)
upper := strings.ToUpper(trimmed)
if strings.HasPrefix(upper, "CREATE TABLE IF NOT EXISTS QUESTION_TABLE_EXPORTS") {
inTable = true
continue
}
if inTable && trimmed == ");" {
inTable = false
continue
}
if !inTable {
continue
}
for _, forbidden := range []string{
"QUESTION", "ANSWER_TEXT", "RAW_PAYLOAD", "SESSIONWEBHOOK", "TOKEN",
"RAW_RESPONSE", "RAW_ERROR", "ACCESS_TOKEN", "APP_SECRET",
} {
if strings.Contains(upper, forbidden) {
t.Errorf("question_table_exports column must not contain %q", forbidden)
}
}
}
}
+19 -10
View File
@@ -179,9 +179,11 @@ func (h *Handler) Handle(ctx context.Context, cb BotCallback) error {
return errors.New("dingtalk handler: sender staffId/senderId is required")
}
senderName := strings.TrimSpace(cb.SenderNick)
triggerPerson := strings.TrimSpace(cb.SenderNick)
senderName := triggerPerson
department := ""
region := ""
actorUnionID := ""
if h.directory != nil {
profile, _ := h.directory.ResolveUser(ctx, h.cfg.TenantKey, externalUserID, senderName)
if name := strings.TrimSpace(profile.Name); name != "" {
@@ -189,18 +191,25 @@ func (h *Handler) Handle(ctx context.Context, cb BotCallback) error {
department = strings.TrimSpace(profile.DepartmentName)
region = strings.TrimSpace(profile.Region)
}
actorUnionID = strings.TrimSpace(profile.UnionID)
}
// Trigger person snapshot falls back to official name when callback nick is blank.
if triggerPerson == "" {
triggerPerson = senderName
}
req := core.QARequest{
TenantKey: h.cfg.TenantKey,
Channel: h.cfg.Channel,
ExternalUserID: externalUserID,
ExternalMessageID: strings.TrimSpace(cb.MsgID),
SenderName: senderName,
Department: department,
Region: region,
MessageType: strings.ToLower(strings.TrimSpace(cb.Msgtype)),
ConversationTitle: strings.TrimSpace(cb.ConversationTitle),
TenantKey: h.cfg.TenantKey,
Channel: h.cfg.Channel,
ExternalUserID: externalUserID,
ExternalMessageID: strings.TrimSpace(cb.MsgID),
SenderName: senderName,
TriggerPersonSnapshot: triggerPerson,
ActorUnionIDSnapshot: actorUnionID,
Department: department,
Region: region,
MessageType: strings.ToLower(strings.TrimSpace(cb.Msgtype)),
ConversationTitle: strings.TrimSpace(cb.ConversationTitle),
}
switch cb.ConversationType {
case conversationTypeGroup:
+133
View File
@@ -222,6 +222,8 @@ func TestHandleDirectUsesDirectoryProfile(t *testing.T) {
}
}
// TestHandleDirectoryFailureFallsBackToSenderNick validates that when directory
// resolution fails, SenderName falls back to the callback SenderNick.
func TestHandleDirectoryFailureFallsBackToSenderNick(t *testing.T) {
qa := &fakeQA{result: core.QAResult{Answer: "ok"}}
reply := &fakeReply{}
@@ -1874,3 +1876,134 @@ func TestHandlePassesConversationTitleToQARequest(t *testing.T) {
}
})
}
// TestActorUnionIDSnapshotPropagation verifies the actor_union_id_snapshot
// is populated from the directory profile UnionID when available.
func TestActorUnionIDSnapshotPropagation(t *testing.T) {
t.Run("resolved profile with union ID captures snapshot", func(t *testing.T) {
qa := &fakeQA{result: core.QAResult{Answer: "ok"}}
reply := &fakeReply{}
profiles := &fakeDirectory{profile: directory.Profile{
Name: "张三", UnionID: "union-abc-123", DepartmentName: "技术部", Region: "APAC",
}}
h, err := NewHandlerWithOptions(TenantConfig{TenantKey: "tenant-a", Channel: "dingtalk"}, qa, reply, HandlerOptions{Directory: profiles})
if err != nil {
t.Fatalf("NewHandlerWithOptions: %v", err)
}
data := directTextCallback()
data.SenderNick = "钉钉昵称"
if err := h.Handle(context.Background(), data); err != nil {
t.Fatalf("Handle: %v", err)
}
if qa.lastReq.ActorUnionIDSnapshot != "union-abc-123" {
t.Fatalf("ActorUnionIDSnapshot = %q, want union-abc-123", qa.lastReq.ActorUnionIDSnapshot)
}
if qa.lastReq.SenderName != "张三" {
t.Fatalf("SenderName = %q, want 张三", qa.lastReq.SenderName)
}
if qa.lastReq.TriggerPersonSnapshot != "钉钉昵称" {
t.Fatalf("TriggerPersonSnapshot = %q, want 钉钉昵称", qa.lastReq.TriggerPersonSnapshot)
}
})
t.Run("blank profile union leaves snapshot empty", func(t *testing.T) {
qa := &fakeQA{result: core.QAResult{Answer: "ok"}}
reply := &fakeReply{}
profiles := &fakeDirectory{profile: directory.Profile{
Name: "张三", UnionID: "", DepartmentName: "技术部", Region: "APAC",
}}
h, err := NewHandlerWithOptions(TenantConfig{TenantKey: "tenant-a", Channel: "dingtalk"}, qa, reply, HandlerOptions{Directory: profiles})
if err != nil {
t.Fatalf("NewHandlerWithOptions: %v", err)
}
data := directTextCallback()
data.SenderNick = "钉钉昵称"
if err := h.Handle(context.Background(), data); err != nil {
t.Fatalf("Handle: %v", err)
}
if qa.lastReq.ActorUnionIDSnapshot != "" {
t.Fatalf("ActorUnionIDSnapshot = %q, want empty", qa.lastReq.ActorUnionIDSnapshot)
}
})
t.Run("directory failure leaves union snapshot empty", func(t *testing.T) {
qa := &fakeQA{result: core.QAResult{Answer: "ok"}}
reply := &fakeReply{}
profiles := &fakeDirectory{err: errors.New("directory unavailable")}
h, err := NewHandlerWithOptions(TenantConfig{TenantKey: "tenant-a", Channel: "dingtalk"}, qa, reply, HandlerOptions{Directory: profiles})
if err != nil {
t.Fatalf("NewHandlerWithOptions: %v", err)
}
data := directTextCallback()
data.SenderNick = "钉钉昵称"
if err := h.Handle(context.Background(), data); err != nil {
t.Fatalf("Handle: %v", err)
}
if qa.lastReq.ActorUnionIDSnapshot != "" {
t.Fatalf("ActorUnionIDSnapshot = %q, want empty on directory failure", qa.lastReq.ActorUnionIDSnapshot)
}
})
t.Run("no directory leaves union snapshot empty", func(t *testing.T) {
qa := &fakeQA{result: core.QAResult{Answer: "ok"}}
reply := &fakeReply{}
h := newTestHandler(t, qa, reply)
data := directTextCallback()
data.SenderNick = "钉钉昵称"
if err := h.Handle(context.Background(), data); err != nil {
t.Fatalf("Handle: %v", err)
}
if qa.lastReq.ActorUnionIDSnapshot != "" {
t.Fatalf("ActorUnionIDSnapshot = %q, want empty without directory", qa.lastReq.ActorUnionIDSnapshot)
}
})
}
// TestTriggerPersonSnapshotRegression proves the trigger_person_snapshot
// contract: the callback SenderNick becomes TriggerPersonSnapshot, the
// directory official Name remains SenderName, and a blank callback nick
// falls back to the official name.
func TestTriggerPersonSnapshotRegression(t *testing.T) {
t.Run("callback nick becomes trigger person snapshot", func(t *testing.T) {
qa := &fakeQA{result: core.QAResult{Answer: "ok"}}
reply := &fakeReply{}
profiles := &fakeDirectory{profile: directory.Profile{Name: "张三", DepartmentName: "技术部", Region: "APAC"}}
h, err := NewHandlerWithOptions(TenantConfig{TenantKey: "tenant-a", Channel: "dingtalk"}, qa, reply, HandlerOptions{Directory: profiles})
if err != nil {
t.Fatalf("NewHandlerWithOptions: %v", err)
}
data := directTextCallback()
data.SenderNick = "钉钉昵称"
if err := h.Handle(context.Background(), data); err != nil {
t.Fatalf("Handle: %v", err)
}
if qa.lastReq.TriggerPersonSnapshot != "钉钉昵称" {
t.Fatalf("TriggerPersonSnapshot = %q, want 钉钉昵称", qa.lastReq.TriggerPersonSnapshot)
}
if qa.lastReq.SenderName != "张三" {
t.Fatalf("SenderName = %q, want 张三", qa.lastReq.SenderName)
}
})
t.Run("blank callback nick falls back to official name", func(t *testing.T) {
qa := &fakeQA{result: core.QAResult{Answer: "ok"}}
reply := &fakeReply{}
profiles := &fakeDirectory{profile: directory.Profile{Name: "张三", DepartmentName: "技术部", Region: "APAC"}}
h, err := NewHandlerWithOptions(TenantConfig{TenantKey: "tenant-a", Channel: "dingtalk"}, qa, reply, HandlerOptions{Directory: profiles})
if err != nil {
t.Fatalf("NewHandlerWithOptions: %v", err)
}
data := directTextCallback()
data.SenderNick = ""
if err := h.Handle(context.Background(), data); err != nil {
t.Fatalf("Handle: %v", err)
}
if qa.lastReq.TriggerPersonSnapshot != "张三" {
t.Fatalf("TriggerPersonSnapshot = %q, want 张三 on blank nick", qa.lastReq.TriggerPersonSnapshot)
}
if qa.lastReq.SenderName != "张三" {
t.Fatalf("SenderName = %q, want 张三", qa.lastReq.SenderName)
}
})
}
+45 -41
View File
@@ -37,19 +37,21 @@ type MediaAudit struct {
}
type QARequest struct {
TenantKey string
Channel string
ConversationType ConversationType
ConversationTitle string
ExternalChatID string
ExternalUserID string
ExternalMessageID string
SenderName string
Department string
Region string
Query string
Images []ImageInput
MessageType string
TenantKey string
Channel string
ConversationType ConversationType
ConversationTitle string
ExternalChatID string
ExternalUserID string
ExternalMessageID string
SenderName string
TriggerPersonSnapshot string
ActorUnionIDSnapshot string
Department string
Region string
Query string
Images []ImageInput
MessageType string
}
type Reference struct {
@@ -86,34 +88,36 @@ type SessionMapping struct {
type SessionTitleUpdate func(ctx context.Context, sessionID, title string) error
type Turn struct {
ID string
TenantKey string
Channel string
ExternalMessageID string
SessionKey string
WeKnoraSessionID string
WeKnoraRequestID string
AgentID string
ConversationType ConversationType
ConversationTitle string
ExternalChatID string
ExternalUserID string
SenderName string
DepartmentSnapshot string
RegionSnapshot string
MessageType string
ActorID string
ConversationID string
RequestMessageID string
AnswerMessageID string
AnswerDeliveryStatus string
AnswerDeliveryMethod string
Question string
Answer string
References []Reference
Media []MediaAudit
Status string
ErrorMessage string
ID string
TenantKey string
Channel string
ExternalMessageID string
SessionKey string
WeKnoraSessionID string
WeKnoraRequestID string
AgentID string
ConversationType ConversationType
ConversationTitle string
ExternalChatID string
ExternalUserID string
SenderName string
TriggerPersonSnapshot string
ActorUnionIDSnapshot string
DepartmentSnapshot string
RegionSnapshot string
MessageType string
ActorID string
ConversationID string
RequestMessageID string
AnswerMessageID string
AnswerDeliveryStatus string
AnswerDeliveryMethod string
Question string
Answer string
References []Reference
Media []MediaAudit
Status string
ErrorMessage string
}
// Delivery status constants for outbound messages.
+22 -20
View File
@@ -136,26 +136,28 @@ func (s *Service) ask(ctx context.Context, request core.QARequest, answerObserve
return core.QAResult{}, errors.New("unsupported message_type")
}
turn := core.Turn{
ID: newID(),
TenantKey: request.TenantKey,
Channel: request.Channel,
ExternalMessageID: request.ExternalMessageID,
SessionKey: sessionKey,
AgentID: agentID,
ConversationType: request.ConversationType,
ConversationTitle: request.ConversationTitle,
ExternalChatID: request.ExternalChatID,
ExternalUserID: request.ExternalUserID,
SenderName: request.SenderName,
DepartmentSnapshot: request.Department,
RegionSnapshot: request.Region,
MessageType: messageType,
ActorID: newID(),
ConversationID: newID(),
RequestMessageID: newID(),
Question: request.Query,
Media: media,
Status: "processing",
ID: newID(),
TenantKey: request.TenantKey,
Channel: request.Channel,
ExternalMessageID: request.ExternalMessageID,
SessionKey: sessionKey,
AgentID: agentID,
ConversationType: request.ConversationType,
ConversationTitle: request.ConversationTitle,
ExternalChatID: request.ExternalChatID,
ExternalUserID: request.ExternalUserID,
SenderName: request.SenderName,
TriggerPersonSnapshot: request.TriggerPersonSnapshot,
ActorUnionIDSnapshot: request.ActorUnionIDSnapshot,
DepartmentSnapshot: request.Department,
RegionSnapshot: request.Region,
MessageType: messageType,
ActorID: newID(),
ConversationID: newID(),
RequestMessageID: newID(),
Question: request.Query,
Media: media,
Status: "processing",
}
stored, created, err := s.repository.BeginTurn(ctx, turn)
if err != nil {
+303
View File
@@ -0,0 +1,303 @@
package questiontable
import (
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
"io"
"net/http"
"net/url"
"strings"
"sync"
"time"
)
// Default DingTalk API base.
const defaultBase = "https://api.dingtalk.com"
// Config holds the required parameters for the DingTalk question-table client.
type Config struct {
ClientID string
Secret string
BaseID string
SheetID string
OperatorID string
// TokenURL overrides the token endpoint. Empty → default.
TokenURL string
// BaseURL overrides the API base. Empty → default.
BaseURL string
}
// defaultTransport is a clone of http.DefaultTransport with Proxy=nil so the
// default client never uses a proxy (even from env). Shared across clients.
var defaultTransport = func() *http.Transport {
dt, ok := http.DefaultTransport.(*http.Transport)
if !ok {
return &http.Transport{
Proxy: nil,
ForceAttemptHTTP2: true,
MaxIdleConns: 100,
IdleConnTimeout: 90 * time.Second,
TLSHandshakeTimeout: 10 * time.Second,
ExpectContinueTimeout: 1 * time.Second,
}
}
// Clone, then clear Proxy.
clone := dt.Clone()
clone.Proxy = nil
return clone
}()
// Client is a stdlib-only DingTalk HTTP client that inserts records into a
// Notable v1 base. It owns a mutex-protected in-memory access-token cache.
type Client struct {
clientID string
secret string
baseID string
sheetID string
operatorID string
tokenURL string
baseURL string
http *http.Client
mu sync.Mutex
token string
tokenExp time.Time
}
// NewClient creates a validated Client. It returns an error when any of
// clientID, secret, baseID, sheetID, or operatorID is empty.
// httpClient may be nil; a default is used in that case.
func NewClient(cfg Config, httpClient *http.Client) (*Client, error) {
// Trim and validate.
clientID := strings.TrimSpace(cfg.ClientID)
if clientID == "" {
return nil, errors.New("questiontable: client id is required")
}
secret := strings.TrimSpace(cfg.Secret)
if secret == "" {
return nil, errors.New("questiontable: secret is required")
}
baseID := strings.TrimSpace(cfg.BaseID)
if baseID == "" {
return nil, errors.New("questiontable: base id is required")
}
sheetID := strings.TrimSpace(cfg.SheetID)
if sheetID == "" {
return nil, errors.New("questiontable: sheet id is required")
}
operatorID := strings.TrimSpace(cfg.OperatorID)
if operatorID == "" {
return nil, errors.New("questiontable: operator id is required")
}
if httpClient == nil {
httpClient = &http.Client{
Transport: defaultTransport,
Timeout: 30 * time.Second,
}
}
tokenURL := strings.TrimSpace(cfg.TokenURL)
if tokenURL == "" {
tokenURL = defaultBase + "/v1.0/oauth2/accessToken"
}
baseURL := strings.TrimSpace(cfg.BaseURL)
if baseURL == "" {
baseURL = defaultBase
}
return &Client{
clientID: clientID,
secret: secret,
baseID: baseID,
sheetID: sheetID,
operatorID: operatorID,
tokenURL: tokenURL,
baseURL: baseURL,
http: httpClient,
}, nil
}
// TokenURL returns the configured token URL (test seam).
func (c *Client) TokenURL() string { return c.tokenURL }
// BaseURL returns the configured API base (test seam).
func (c *Client) BaseURL() string { return c.baseURL }
// insertRecordReq is the JSON body for the Notable v1 insertRecord endpoint.
type insertRecordReq struct {
Records []map[string]interface{} `json:"records"`
}
// insertRecordResp is the success response body.
type insertRecordResp struct {
Value []struct {
ID string `json:"id"`
} `json:"value"`
}
// tokenResp is the access-token response body.
type tokenResp struct {
AccessToken string `json:"accessToken"`
ExpireIn int64 `json:"expireIn"`
}
// InsertRecord sends a single row to the DingTalk Notable v1 base.
// It returns the record ID on success.
func (c *Client) InsertRecord(ctx context.Context, row Row) (string, error) {
token, err := c.getAccessToken(ctx)
if err != nil {
return "", fmt.Errorf("questiontable token: %w", err)
}
path := fmt.Sprintf("/v1.0/notable/bases/%s/sheets/%s/records",
url.PathEscape(c.baseID), url.PathEscape(c.sheetID))
reqURL := c.baseURL + path + "?operatorId=" + url.QueryEscape(c.operatorID)
fields := row.ToFields()
body := insertRecordReq{Records: []map[string]interface{}{fields}}
jsonBody, err := json.Marshal(body)
if err != nil {
return "", fmt.Errorf("questiontable marshal: %w", err)
}
req, err := http.NewRequestWithContext(ctx, http.MethodPost, reqURL, bytes.NewReader(jsonBody))
if err != nil {
return "", fmt.Errorf("questiontable request: %w", err)
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("x-acs-dingtalk-access-token", token)
resp, err := c.http.Do(req)
if err != nil {
return "", safeDoErr(ctx, err)
}
defer resp.Body.Close()
// Read bounded response body.
raw, err := io.ReadAll(io.LimitReader(resp.Body, 64<<10))
if err != nil {
return "", fmt.Errorf("questiontable read response: %w", err)
}
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
return "", fmt.Errorf("questiontable insert returned HTTP %d", resp.StatusCode)
}
var ir insertRecordResp
if err := json.Unmarshal(raw, &ir); err != nil {
return "", fmt.Errorf("questiontable decode response: %w", err)
}
if len(ir.Value) == 0 {
return "", errors.New("questiontable empty value list")
}
if strings.TrimSpace(ir.Value[0].ID) == "" {
return "", errors.New("questiontable empty record id")
}
return ir.Value[0].ID, nil
}
// safeDoErr wraps an http.Do error without exposing the raw URL or credentials.
// If the context is already done, returns a wrapped ctx.Err() so errors.Is works.
func safeDoErr(ctx context.Context, err error) error {
if ctx.Err() != nil {
return fmt.Errorf("questiontable insert: %w", ctx.Err())
}
return errors.New("questiontable insert: network error")
}
// getAccessToken returns a cached token or fetches a new one.
func (c *Client) getAccessToken(ctx context.Context) (string, error) {
c.mu.Lock()
if c.token != "" && time.Now().Before(c.tokenExp) {
tok := c.token
c.mu.Unlock()
return tok, nil
}
c.mu.Unlock()
return c.fetchToken(ctx)
}
// fetchToken calls the DingTalk OAuth2 endpoint and caches the result.
func (c *Client) fetchToken(ctx context.Context) (string, error) {
c.mu.Lock()
defer c.mu.Unlock()
// Double-check inside lock.
if c.token != "" && time.Now().Before(c.tokenExp) {
return c.token, nil
}
body := map[string]string{
"appKey": c.clientID,
"appSecret": c.secret,
}
jsonBody, err := json.Marshal(body)
if err != nil {
return "", fmt.Errorf("questiontable token marshal: %w", err)
}
req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.tokenURL, bytes.NewReader(jsonBody))
if err != nil {
return "", fmt.Errorf("questiontable token request: %w", err)
}
req.Header.Set("Content-Type", "application/json")
resp, err := c.http.Do(req)
if err != nil {
return "", safeTokenDoErr(ctx, err)
}
defer resp.Body.Close()
raw, err := io.ReadAll(io.LimitReader(resp.Body, 64<<10))
if err != nil {
return "", fmt.Errorf("questiontable token read: %w", err)
}
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
return "", fmt.Errorf("questiontable token returned HTTP %d", resp.StatusCode)
}
var tr tokenResp
if err := json.Unmarshal(raw, &tr); err != nil {
return "", fmt.Errorf("questiontable token decode: %w", err)
}
if strings.TrimSpace(tr.AccessToken) == "" {
return "", errors.New("questiontable empty access token")
}
// Token expiry must never be extended beyond DingTalk's expireIn.
if tr.ExpireIn <= 0 {
return "", errors.New("questiontable token: non-positive expireIn")
}
exp := time.Duration(tr.ExpireIn) * time.Second
// For very short TTL (<=30s), cache for a positive fraction (TTL/2).
if exp <= 30*time.Second {
exp = exp / 2
} else {
// Safety margin: subtract 30 seconds from expiry.
exp -= 30 * time.Second
}
c.token = tr.AccessToken
c.tokenExp = time.Now().Add(exp)
return c.token, nil
}
// safeTokenDoErr wraps an http.Do error from token fetch without exposing raw URL or credentials.
func safeTokenDoErr(ctx context.Context, err error) error {
if ctx.Err() != nil {
return fmt.Errorf("questiontable token: %w", ctx.Err())
}
return errors.New("questiontable token: network error")
}
+559
View File
@@ -0,0 +1,559 @@
package questiontable
import (
"context"
"encoding/json"
"errors"
"net/http"
"net/http/httptest"
"strings"
"sync/atomic"
"testing"
"time"
)
func TestNewClientRejectsMissingConfig(t *testing.T) {
tests := []struct {
name string
cfg Config
}{
{"empty client id", Config{Secret: "s", BaseID: "b", SheetID: "sh", OperatorID: "op"}},
{"empty secret", Config{ClientID: "c", BaseID: "b", SheetID: "sh", OperatorID: "op"}},
{"empty base id", Config{ClientID: "c", Secret: "s", SheetID: "sh", OperatorID: "op"}},
{"empty sheet id", Config{ClientID: "c", Secret: "s", BaseID: "b", OperatorID: "op"}},
{"empty operator id", Config{ClientID: "c", Secret: "s", BaseID: "b", SheetID: "sh"}},
}
for _, tt := range tests {
_, err := NewClient(tt.cfg, nil)
if err == nil {
t.Errorf("%s: expected error, got nil", tt.name)
}
}
}
func TestInsertRecordMethodPathHeaderBody(t *testing.T) {
var (
gotMethod string
gotPath string
gotQuery string
gotToken string
gotBody string
tokenCalls int32
insertCalls int32
)
mux := http.NewServeMux()
mux.HandleFunc("/v1.0/oauth2/accessToken", func(w http.ResponseWriter, r *http.Request) {
atomic.AddInt32(&tokenCalls, 1)
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write([]byte(`{"accessToken":"test-access-token","expireIn":7200}`))
})
mux.HandleFunc("/v1.0/notable/bases/b1/sheets/sh1/records", func(w http.ResponseWriter, r *http.Request) {
atomic.AddInt32(&insertCalls, 1)
gotMethod = r.Method
gotPath = r.URL.Path
gotQuery = r.URL.RawQuery
gotToken = r.Header.Get("x-acs-dingtalk-access-token")
var body insertRecordReq
if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
t.Errorf("decode body: %v", err)
}
raw, _ := json.Marshal(body)
gotBody = string(raw)
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write([]byte(`{"value":[{"id":"rec-abc-123"}]}`))
})
server := httptest.NewServer(mux)
defer server.Close()
client, err := NewClient(Config{
ClientID: "appKey-1",
Secret: "appSecret-1",
BaseID: "b1",
SheetID: "sh1",
OperatorID: "operator-1",
TokenURL: server.URL + "/v1.0/oauth2/accessToken",
BaseURL: server.URL,
}, nil)
if err != nil {
t.Fatalf("NewClient: %v", err)
}
row := Row{
Question: "测试问题",
Answer: "测试答案",
AskerName: "张三",
AskerUnionID: "union-zhang-san",
}
recID, err := client.InsertRecord(context.Background(), row)
if err != nil {
t.Fatalf("InsertRecord: %v", err)
}
if recID != "rec-abc-123" {
t.Errorf("recordID = %q, want %q", recID, "rec-abc-123")
}
// Assert method, path, header, token.
if gotMethod != http.MethodPost {
t.Errorf("method = %q, want POST", gotMethod)
}
if gotPath != "/v1.0/notable/bases/b1/sheets/sh1/records" {
t.Errorf("path = %q", gotPath)
}
if gotToken != "test-access-token" {
t.Errorf("token header = %q, want test-access-token", gotToken)
}
// operatorId must be in query only.
if !strings.Contains(gotQuery, "operatorId=operator-1") {
t.Errorf("query missing operatorId: %q", gotQuery)
}
// body must NOT contain operatorId.
if strings.Contains(gotBody, "operatorId") {
t.Errorf("body contains operatorId: %q", gotBody)
}
// body must contain typed field IDs.
if !strings.Contains(gotBody, `"ZaMZF6j"`) {
t.Errorf("body missing field ID ZaMZF6j: %q", gotBody)
}
if !strings.Contains(gotBody, `"Y4cpmxU"`) {
t.Errorf("body missing field ID Y4cpmxU: %q", gotBody)
}
// body must NOT contain the omitted select fields.
omitted := []string{"QMn9UoU", "H3gtTuv", "nFvv8OR", "UMqUPAG", "lnF7gF9"}
for _, fid := range omitted {
if strings.Contains(gotBody, fid) {
t.Errorf("body contains omitted select field %q: %q", fid, gotBody)
}
}
}
func TestTokenCacheTwoInsertsOneTokenCall(t *testing.T) {
var tokenCalls int32
mux := http.NewServeMux()
mux.HandleFunc("/v1.0/oauth2/accessToken", func(w http.ResponseWriter, r *http.Request) {
atomic.AddInt32(&tokenCalls, 1)
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write([]byte(`{"accessToken":"cached-token","expireIn":7200}`))
})
mux.HandleFunc("/v1.0/notable/bases/b1/sheets/sh1/records", func(w http.ResponseWriter, r *http.Request) {
if r.Header.Get("x-acs-dingtalk-access-token") != "cached-token" {
t.Errorf("unexpected token on insert")
}
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write([]byte(`{"value":[{"id":"rec-1"}]}`))
})
server := httptest.NewServer(mux)
defer server.Close()
client, err := NewClient(Config{
ClientID: "appKey-1",
Secret: "appSecret-1",
BaseID: "b1",
SheetID: "sh1",
OperatorID: "operator-1",
TokenURL: server.URL + "/v1.0/oauth2/accessToken",
BaseURL: server.URL,
}, nil)
if err != nil {
t.Fatalf("NewClient: %v", err)
}
row := Row{Question: "q", Answer: "a", AskerName: "x", AskerUnionID: "union-x"}
// Two inserts, only one token call.
for i := 0; i < 2; i++ {
if _, err := client.InsertRecord(context.Background(), row); err != nil {
t.Fatalf("InsertRecord %d: %v", i, err)
}
}
if tokenCalls != 1 {
t.Errorf("token calls = %d, want 1", tokenCalls)
}
}
func TestInsertRecordNon2xx(t *testing.T) {
mux := http.NewServeMux()
mux.HandleFunc("/v1.0/oauth2/accessToken", func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write([]byte(`{"accessToken":"tok","expireIn":7200}`))
})
mux.HandleFunc("/v1.0/notable/bases/b1/sheets/sh1/records", func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusBadRequest)
_, _ = w.Write([]byte(`{"code":"invalid","message":"bad request"}`))
})
server := httptest.NewServer(mux)
defer server.Close()
client, _ := NewClient(Config{
ClientID: "c", Secret: "s", BaseID: "b1", SheetID: "sh1", OperatorID: "op",
TokenURL: server.URL + "/v1.0/oauth2/accessToken", BaseURL: server.URL,
}, nil)
row := Row{Question: "q", Answer: "a", AskerName: "x", AskerUnionID: "union-x"}
_, err := client.InsertRecord(context.Background(), row)
if err == nil {
t.Fatal("expected error for non-2xx")
}
// Error must not contain token, secret, operator, or raw row values.
errStr := err.Error()
for _, secret := range []string{"tok", "op", "union-x"} {
if strings.Contains(errStr, secret) {
t.Errorf("error redaction failed: %q found in %q", secret, errStr)
}
}
}
func TestInsertRecordMalformedResponse(t *testing.T) {
mux := http.NewServeMux()
mux.HandleFunc("/v1.0/oauth2/accessToken", func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write([]byte(`{"accessToken":"tok","expireIn":7200}`))
})
mux.HandleFunc("/v1.0/notable/bases/b1/sheets/sh1/records", func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write([]byte(`not json`))
})
server := httptest.NewServer(mux)
defer server.Close()
client, _ := NewClient(Config{
ClientID: "c", Secret: "s", BaseID: "b1", SheetID: "sh1", OperatorID: "op",
TokenURL: server.URL + "/v1.0/oauth2/accessToken", BaseURL: server.URL,
}, nil)
row := Row{Question: "q", Answer: "a", AskerName: "x", AskerUnionID: "union-x"}
_, err := client.InsertRecord(context.Background(), row)
if err == nil {
t.Fatal("expected error for malformed response")
}
}
func TestInsertRecordEmptyValueList(t *testing.T) {
mux := http.NewServeMux()
mux.HandleFunc("/v1.0/oauth2/accessToken", func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write([]byte(`{"accessToken":"tok","expireIn":7200}`))
})
mux.HandleFunc("/v1.0/notable/bases/b1/sheets/sh1/records", func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write([]byte(`{"value":[]}`))
})
server := httptest.NewServer(mux)
defer server.Close()
client, _ := NewClient(Config{
ClientID: "c", Secret: "s", BaseID: "b1", SheetID: "sh1", OperatorID: "op",
TokenURL: server.URL + "/v1.0/oauth2/accessToken", BaseURL: server.URL,
}, nil)
row := Row{Question: "q", Answer: "a", AskerName: "x", AskerUnionID: "union-x"}
_, err := client.InsertRecord(context.Background(), row)
if err == nil {
t.Fatal("expected error for empty value list")
}
}
func TestInsertRecordEmptyRecordID(t *testing.T) {
mux := http.NewServeMux()
mux.HandleFunc("/v1.0/oauth2/accessToken", func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write([]byte(`{"accessToken":"tok","expireIn":7200}`))
})
mux.HandleFunc("/v1.0/notable/bases/b1/sheets/sh1/records", func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write([]byte(`{"value":[{"id":""}]}`))
})
server := httptest.NewServer(mux)
defer server.Close()
client, _ := NewClient(Config{
ClientID: "c", Secret: "s", BaseID: "b1", SheetID: "sh1", OperatorID: "op",
TokenURL: server.URL + "/v1.0/oauth2/accessToken", BaseURL: server.URL,
}, nil)
row := Row{Question: "q", Answer: "a", AskerName: "x", AskerUnionID: "union-x"}
_, err := client.InsertRecord(context.Background(), row)
if err == nil {
t.Fatal("expected error for empty record id")
}
}
func TestTokenCallErrorRedaction(t *testing.T) {
// Token endpoint returns non-2xx.
mux := http.NewServeMux()
mux.HandleFunc("/v1.0/oauth2/accessToken", func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusUnauthorized)
_, _ = w.Write([]byte(`{"code":"unauthorized","message":"invalid appSecret"}`))
})
server := httptest.NewServer(mux)
defer server.Close()
client, _ := NewClient(Config{
ClientID: "c", Secret: "s", BaseID: "b1", SheetID: "sh1", OperatorID: "op",
TokenURL: server.URL + "/v1.0/oauth2/accessToken", BaseURL: server.URL,
}, nil)
row := Row{Question: "q", Answer: "a", AskerName: "x", AskerUnionID: "union-x"}
_, err := client.InsertRecord(context.Background(), row)
if err == nil {
t.Fatal("expected error for token failure")
}
// Error must not contain raw body, secret, or token.
errStr := err.Error()
for _, secret := range []string{"appSecret", "invalid appSecret", "unauthorized"} {
if strings.Contains(errStr, secret) {
t.Errorf("error redaction failed: %q found in %q", secret, errStr)
}
}
}
func TestTokenEmptyAccessToken(t *testing.T) {
mux := http.NewServeMux()
mux.HandleFunc("/v1.0/oauth2/accessToken", func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write([]byte(`{"accessToken":"","expireIn":7200}`))
})
server := httptest.NewServer(mux)
defer server.Close()
client, _ := NewClient(Config{
ClientID: "c", Secret: "s", BaseID: "b1", SheetID: "sh1", OperatorID: "op",
TokenURL: server.URL + "/v1.0/oauth2/accessToken", BaseURL: server.URL,
}, nil)
row := Row{Question: "q", Answer: "a", AskerName: "x", AskerUnionID: "union-x"}
_, err := client.InsertRecord(context.Background(), row)
if err == nil {
t.Fatal("expected error for empty access token")
}
}
func TestTokenExpirySafetyMargin(t *testing.T) {
// Token with expireIn=1: TTL ≤ 30s → exp = TTL/2 (500ms). The tokenExp
// must be after now but no later than server expiry (1s) + small scheduling
// tolerance. An immediate second request should still reuse the cache.
var tokenCalls int32
mux := http.NewServeMux()
mux.HandleFunc("/v1.0/oauth2/accessToken", func(w http.ResponseWriter, r *http.Request) {
atomic.AddInt32(&tokenCalls, 1)
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write([]byte(`{"accessToken":"short-token","expireIn":1}`))
})
mux.HandleFunc("/v1.0/notable/bases/b1/sheets/sh1/records", func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write([]byte(`{"value":[{"id":"rec-1"}]}`))
})
server := httptest.NewServer(mux)
defer server.Close()
client, _ := NewClient(Config{
ClientID: "c", Secret: "s", BaseID: "b1", SheetID: "sh1", OperatorID: "op",
TokenURL: server.URL + "/v1.0/oauth2/accessToken", BaseURL: server.URL,
}, nil)
row := Row{Question: "q", Answer: "a", AskerName: "x", AskerUnionID: "union-x"}
// First call: token fetch.
before := time.Now()
_, err := client.InsertRecord(context.Background(), row)
if err != nil {
t.Fatalf("first InsertRecord: %v", err)
}
if tokenCalls != 1 {
t.Errorf("token calls after first insert = %d, want 1", tokenCalls)
}
// Inspect tokenExp.
client.mu.Lock()
exp := client.tokenExp
client.mu.Unlock()
// tokenExp must be after now.
if !exp.After(before) {
t.Errorf("tokenExp %v is not after request start %v", exp, before)
}
// tokenExp must not exceed server expiry (1s) + scheduling tolerance (500ms).
if exp.After(before.Add(1500 * time.Millisecond)) {
t.Errorf("tokenExp %v exceeds server expiry + tolerance", exp)
}
// Second call: should use cached token (within TTL/2 = 500ms).
_, err = client.InsertRecord(context.Background(), row)
if err != nil {
t.Fatalf("second InsertRecord: %v", err)
}
if tokenCalls != 1 {
t.Errorf("token calls after second insert = %d, want 1 (cached)", tokenCalls)
}
}
func TestTokenExpiryNonPositive(t *testing.T) {
mux := http.NewServeMux()
mux.HandleFunc("/v1.0/oauth2/accessToken", func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write([]byte(`{"accessToken":"tok","expireIn":0}`))
})
server := httptest.NewServer(mux)
defer server.Close()
client, _ := NewClient(Config{
ClientID: "c", Secret: "s", BaseID: "b1", SheetID: "sh1", OperatorID: "op",
TokenURL: server.URL + "/v1.0/oauth2/accessToken", BaseURL: server.URL,
}, nil)
row := Row{Question: "q", Answer: "a", AskerName: "x", AskerUnionID: "union-x"}
_, err := client.InsertRecord(context.Background(), row)
if err == nil {
t.Fatal("expected error for non-positive expireIn")
}
}
func TestContextTimeout(t *testing.T) {
handlerDone := make(chan struct{})
mux := http.NewServeMux()
mux.HandleFunc("/v1.0/oauth2/accessToken", func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write([]byte(`{"accessToken":"tok","expireIn":7200}`))
})
mux.HandleFunc("/v1.0/notable/bases/b1/sheets/sh1/records", func(w http.ResponseWriter, r *http.Request) {
// Wait for context cancellation or test signal.
select {
case <-r.Context().Done():
case <-handlerDone:
}
})
server := httptest.NewServer(mux)
defer server.Close()
defer close(handlerDone) // runs before server.Close (LIFO), unblocks the handler
// Use a client with keep-alives disabled so the TCP connection
// closes after the handler returns — httptest.Server.Close won't hang.
httpClient := &http.Client{
Transport: &http.Transport{DisableKeepAlives: true},
Timeout: 30 * time.Second,
}
client, _ := NewClient(Config{
ClientID: "c", Secret: "s", BaseID: "b1", SheetID: "sh1", OperatorID: "op",
TokenURL: server.URL + "/v1.0/oauth2/accessToken", BaseURL: server.URL,
}, httpClient)
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond)
defer cancel()
row := Row{Question: "q", Answer: "a", AskerName: "x", AskerUnionID: "union-x"}
_, err := client.InsertRecord(ctx, row)
if err == nil {
t.Fatal("expected timeout error")
}
if !errors.Is(err, context.DeadlineExceeded) && !strings.Contains(err.Error(), "context deadline") {
t.Logf("timeout error (may be wrapped): %v", err)
}
}
func TestDefaultURLs(t *testing.T) {
client, err := NewClient(Config{
ClientID: "c", Secret: "s", BaseID: "b1", SheetID: "sh1", OperatorID: "op",
}, nil)
if err != nil {
t.Fatalf("NewClient: %v", err)
}
if client.TokenURL() != defaultBase+"/v1.0/oauth2/accessToken" {
t.Errorf("default token URL = %q, want %q", client.TokenURL(), defaultBase+"/v1.0/oauth2/accessToken")
}
if client.BaseURL() != defaultBase {
t.Errorf("default base URL = %q, want %q", client.BaseURL(), defaultBase)
}
}
func TestCustomURLs(t *testing.T) {
client, err := NewClient(Config{
ClientID: "c", Secret: "s", BaseID: "b1", SheetID: "sh1", OperatorID: "op",
TokenURL: "https://custom.example.com/token",
BaseURL: "https://custom.example.com",
}, nil)
if err != nil {
t.Fatalf("NewClient: %v", err)
}
if client.TokenURL() != "https://custom.example.com/token" {
t.Errorf("custom token URL = %q", client.TokenURL())
}
if client.BaseURL() != "https://custom.example.com" {
t.Errorf("custom base URL = %q", client.BaseURL())
}
}
func TestNewClientNilHTTP(t *testing.T) {
client, err := NewClient(Config{
ClientID: "c", Secret: "s", BaseID: "b1", SheetID: "sh1", OperatorID: "op",
}, nil)
if err != nil {
t.Fatalf("NewClient with nil http: %v", err)
}
if client.http == nil {
t.Fatal("http client is nil after construction with nil argument")
}
}
func TestNewClientCustomHTTP(t *testing.T) {
custom := &http.Client{Timeout: 5 * time.Second}
client, err := NewClient(Config{
ClientID: "c", Secret: "s", BaseID: "b1", SheetID: "sh1", OperatorID: "op",
}, custom)
if err != nil {
t.Fatalf("NewClient with custom http: %v", err)
}
if client.http != custom {
t.Fatal("http client is not the custom one")
}
}
func TestDefaultTransportNoProxy(t *testing.T) {
// The default transport must have Proxy=nil.
if defaultTransport.Proxy != nil {
t.Error("defaultTransport Proxy is not nil")
}
}
func TestInsertRecordDoErrorRedaction(t *testing.T) {
// Custom RoundTripper that returns an error containing secrets.
secretErr := errors.New("connect: dial tcp https://api.dingtalk.com?operatorId=secret-op failed: row data union-x")
rt := &errorRoundTripper{err: secretErr}
client, _ := NewClient(Config{
ClientID: "c", Secret: "s", BaseID: "b1", SheetID: "sh1", OperatorID: "op",
TokenURL: "https://token.example.com", BaseURL: "https://api.example.com",
}, &http.Client{Transport: rt})
row := Row{Question: "q", Answer: "a", AskerName: "x", AskerUnionID: "union-x"}
_, err := client.InsertRecord(context.Background(), row)
if err == nil {
t.Fatal("expected error")
}
errStr := err.Error()
for _, secret := range []string{"secret-op", "union-x", "api.dingtalk.com"} {
if strings.Contains(strings.ToLower(errStr), strings.ToLower(secret)) {
t.Errorf("error redaction failed: %q found in %q", secret, errStr)
}
}
}
// errorRoundTripper returns a fixed error on every RoundTrip.
type errorRoundTripper struct {
err error
}
func (rt *errorRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) {
return nil, rt.err
}
+191
View File
@@ -0,0 +1,191 @@
package questiontable
import (
"context"
"errors"
"fmt"
"reflect"
"time"
)
// FailureCodeRemoteInsertFailed is the stable public failure code used when
// InsertRecord returns an error. The raw error is never persisted or returned.
const FailureCodeRemoteInsertFailed = "remote_question_table_insert_failed"
// PendingExport is a row waiting to be exported to the remote question table.
type PendingExport struct {
ExportID string
Row Row
Attempts int
}
// ExportRepository is the persistence contract for the background dispatcher.
type ExportRepository interface {
// Claim returns up to limit pending exports whose attempt count is below
// maxAttempts and whose last claim time is before staleBefore.
Claim(ctx context.Context, limit int, maxAttempts int, staleBefore time.Time) ([]PendingExport, error)
// MarkExported records the location (Notable record ID) for a successfully exported row.
MarkExported(ctx context.Context, exportID, location string) error
// MarkFailed records a failure for the given exportID with a stable failure code.
MarkFailed(ctx context.Context, exportID, failureCode string) error
}
// RecordInserter is the contract for inserting a row to the remote Notable base.
// *Client.InsertRecord satisfies this interface.
type RecordInserter interface {
InsertRecord(ctx context.Context, row Row) (string, error)
}
// DispatcherStats holds the outcome counters for a single RunOnce invocation.
type DispatcherStats struct {
Claimed int
Exported int
Failed int
}
// DispatcherOptions holds optional configuration for the Dispatcher.
type DispatcherOptions struct {
Batch int
MaxAttempts int
ClaimLease time.Duration
SendTimeout time.Duration
}
// Dispatcher runs the background dispatch loop that claims pending exports
// and inserts them to the remote question table.
type Dispatcher struct {
repo ExportRepository
inserter RecordInserter
batch int
maxAttempts int
claimLease time.Duration
sendTimeout time.Duration
}
// isNil reports whether x is nil. It accepts any value and only calls
// IsNil when the underlying kind is nil-able. Value types (structs, ints,
// strings, etc.) are never nil and return false without panicking.
func isNil(x interface{}) bool {
if x == nil {
return true
}
v := reflect.ValueOf(x)
switch v.Kind() {
case reflect.Chan, reflect.Func, reflect.Interface, reflect.Map, reflect.Ptr, reflect.Slice:
return v.IsNil()
default:
return false
}
}
// NewDispatcher creates a validated Dispatcher. It returns an error when repo
// or inserter is nil (including typed nil), or when any option value is out of
// bounds. Defaults: batch=10, maxAttempts=5, claimLease=5m, sendTimeout=10s.
func NewDispatcher(repo ExportRepository, inserter RecordInserter, opts *DispatcherOptions) (*Dispatcher, error) {
if isNil(repo) {
return nil, errors.New("questiontable dispatcher: repo is nil")
}
if isNil(inserter) {
return nil, errors.New("questiontable dispatcher: inserter is nil")
}
batch := 10
maxAttempts := 5
claimLease := 5 * time.Minute
sendTimeout := 10 * time.Second
if opts != nil {
if opts.Batch < 0 {
return nil, fmt.Errorf("questiontable dispatcher: batch must be >= 0, got %d", opts.Batch)
}
if opts.Batch > 0 {
batch = opts.Batch
}
if opts.MaxAttempts < 0 {
return nil, fmt.Errorf("questiontable dispatcher: maxAttempts must be >= 0, got %d", opts.MaxAttempts)
}
if opts.MaxAttempts > 0 {
maxAttempts = opts.MaxAttempts
}
if opts.ClaimLease < 0 {
return nil, fmt.Errorf("questiontable dispatcher: claimLease must be >= 0, got %v", opts.ClaimLease)
}
if opts.ClaimLease > 0 {
claimLease = opts.ClaimLease
}
if opts.SendTimeout < 0 {
return nil, fmt.Errorf("questiontable dispatcher: sendTimeout must be >= 0, got %v", opts.SendTimeout)
}
if opts.SendTimeout > 0 {
sendTimeout = opts.SendTimeout
}
}
return &Dispatcher{
repo: repo,
inserter: inserter,
batch: batch,
maxAttempts: maxAttempts,
claimLease: claimLease,
sendTimeout: sendTimeout,
}, nil
}
// RunOnce claims pending exports and dispatches them to the remote table.
// It returns stats and any error that prevented the entire operation.
// Individual row failures are tracked in stats.Failed and do not cause
// RunOnce to return an error.
func (d *Dispatcher) RunOnce(ctx context.Context) (DispatcherStats, error) {
var stats DispatcherStats
// Parent context precheck.
if err := ctx.Err(); err != nil {
return stats, fmt.Errorf("questiontable dispatcher: context invalid: %w", err)
}
staleBefore := time.Now().Add(-d.claimLease)
items, err := d.repo.Claim(ctx, d.batch, d.maxAttempts, staleBefore)
if err != nil {
// Repository errors are not trusted; return stable stage-only error.
return stats, errors.New("questiontable dispatcher: claim failed")
}
stats.Claimed = len(items)
for _, item := range items {
// Check parent context before each item.
if err := ctx.Err(); err != nil {
return stats, fmt.Errorf("questiontable dispatcher: context cancelled during dispatch: %w", err)
}
sendCtx, cancel := context.WithTimeout(ctx, d.sendTimeout)
recordID, insertErr := d.inserter.InsertRecord(sendCtx, item.Row)
cancel()
if insertErr != nil {
// Mark failed with stable code only; raw error not persisted.
if markErr := d.repo.MarkFailed(ctx, item.ExportID, FailureCodeRemoteInsertFailed); markErr != nil {
// Repository errors are not trusted; return stable stage-only error.
return stats, errors.New("questiontable dispatcher: mark failed failed")
}
stats.Failed++
continue
}
// Mark exported with returned Notable record ID.
if markErr := d.repo.MarkExported(ctx, item.ExportID, recordID); markErr != nil {
// Repository errors are not trusted; return stable stage-only error.
return stats, errors.New("questiontable dispatcher: mark exported failed")
}
stats.Exported++
}
return stats, nil
}
+926
View File
@@ -0,0 +1,926 @@
package questiontable
import (
"context"
"errors"
"fmt"
"strings"
"sync"
"testing"
"time"
)
// ---------------------------------------------------------------------------
// Test doubles
// ---------------------------------------------------------------------------
// stubRepo implements ExportRepository for tests.
type stubRepo struct {
mu sync.Mutex
claimFn func(ctx context.Context, limit int, maxAttempts int, staleBefore time.Time) ([]PendingExport, error)
// captured params from last Claim call.
lastClaimLimit int
lastClaimMaxAttempts int
lastClaimStaleBefore time.Time
// exported and failed records for assertions.
exported []exportedRecord
failed []failedRecord
// markExportedErr is returned by MarkExported when non-nil.
markExportedErr error
// markFailedErr is returned by MarkFailed when non-nil.
markFailedErr error
}
type exportedRecord struct {
ExportID string
Location string
}
type failedRecord struct {
ExportID string
FailureCode string
}
func newStubRepo() *stubRepo {
return &stubRepo{}
}
func (r *stubRepo) Claim(ctx context.Context, limit int, maxAttempts int, staleBefore time.Time) ([]PendingExport, error) {
r.mu.Lock()
r.lastClaimLimit = limit
r.lastClaimMaxAttempts = maxAttempts
r.lastClaimStaleBefore = staleBefore
r.mu.Unlock()
if r.claimFn != nil {
return r.claimFn(ctx, limit, maxAttempts, staleBefore)
}
return nil, nil
}
func (r *stubRepo) MarkExported(ctx context.Context, exportID, location string) error {
r.mu.Lock()
defer r.mu.Unlock()
r.exported = append(r.exported, exportedRecord{ExportID: exportID, Location: location})
if r.markExportedErr != nil {
return r.markExportedErr
}
return nil
}
func (r *stubRepo) MarkFailed(ctx context.Context, exportID, failureCode string) error {
r.mu.Lock()
defer r.mu.Unlock()
r.failed = append(r.failed, failedRecord{ExportID: exportID, FailureCode: failureCode})
if r.markFailedErr != nil {
return r.markFailedErr
}
return nil
}
// stubInserter implements RecordInserter for tests.
type stubInserter struct {
mu sync.Mutex
rows []Row
recordID string
err error
blockCh chan struct{} // if set, blocks until closed
callCount int
// perCallErrors maps call index (1-based) to error to return.
// If an entry exists for a call index, that error is returned instead of a.err.
perCallErrors map[int]error
}
func newStubInserter() *stubInserter {
return &stubInserter{recordID: "rec-test-001"}
}
func (a *stubInserter) InsertRecord(ctx context.Context, row Row) (string, error) {
a.mu.Lock()
a.callCount++
call := a.callCount
a.rows = append(a.rows, row)
blockCh := a.blockCh
a.mu.Unlock()
if blockCh != nil {
select {
case <-ctx.Done():
return "", ctx.Err()
case <-blockCh:
}
}
a.mu.Lock()
defer a.mu.Unlock()
// Check per-call error override.
if a.perCallErrors != nil {
if e, ok := a.perCallErrors[call]; ok {
return a.recordID, e
}
}
return a.recordID, a.err
}
// ---------------------------------------------------------------------------
// Test 1: empty queue
// ---------------------------------------------------------------------------
func TestDispatcherRunOnceEmptyQueue(t *testing.T) {
repo := newStubRepo()
inserter := newStubInserter()
d, err := NewDispatcher(repo, inserter, nil)
if err != nil {
t.Fatalf("NewDispatcher: %v", err)
}
stats, err := d.RunOnce(context.Background())
if err != nil {
t.Fatalf("RunOnce: %v", err)
}
if stats.Claimed != 0 {
t.Errorf("Claimed = %d, want 0", stats.Claimed)
}
if stats.Exported != 0 {
t.Errorf("Exported = %d, want 0", stats.Exported)
}
if stats.Failed != 0 {
t.Errorf("Failed = %d, want 0", stats.Failed)
}
}
// ---------------------------------------------------------------------------
// Test 2: success and location mark
// ---------------------------------------------------------------------------
func TestDispatcherRunOnceSuccess(t *testing.T) {
repo := newStubRepo()
inserter := newStubInserter()
inserter.recordID = "rec-notable-abc-123"
exportID := "exp-001"
repo.claimFn = func(ctx context.Context, limit int, maxAttempts int, staleBefore time.Time) ([]PendingExport, error) {
return []PendingExport{
{ExportID: exportID, Row: Row{Question: "hello", Answer: "world", AskerName: "alice", AskerUnionID: "union-alice"}, Attempts: 0},
}, nil
}
d, err := NewDispatcher(repo, inserter, nil)
if err != nil {
t.Fatalf("NewDispatcher: %v", err)
}
stats, err := d.RunOnce(context.Background())
if err != nil {
t.Fatalf("RunOnce: %v", err)
}
if stats.Claimed != 1 {
t.Errorf("Claimed = %d, want 1", stats.Claimed)
}
if stats.Exported != 1 {
t.Errorf("Exported = %d, want 1", stats.Exported)
}
if stats.Failed != 0 {
t.Errorf("Failed = %d, want 0", stats.Failed)
}
// Check location was marked with the Notable record ID.
repo.mu.Lock()
defer repo.mu.Unlock()
if len(repo.exported) != 1 {
t.Fatalf("exported records = %d, want 1", len(repo.exported))
}
if repo.exported[0].ExportID != exportID {
t.Errorf("exported exportID = %q, want %q", repo.exported[0].ExportID, exportID)
}
if repo.exported[0].Location != "rec-notable-abc-123" {
t.Errorf("exported Location = %q, want %q", repo.exported[0].Location, "rec-notable-abc-123")
}
}
// ---------------------------------------------------------------------------
// Test 3: insert error => stable failure code, raw error absent
// ---------------------------------------------------------------------------
func TestDispatcherRunOnceInsertErrorStableCode(t *testing.T) {
repo := newStubRepo()
inserter := newStubInserter()
// Use an error that contains "secrets" that must not leak.
inserter.err = fmt.Errorf("network: connect to https://api.dingtalk.com?operatorId=secret-operator failed: row contains secret-data-xyz")
exportID := "exp-002"
repo.claimFn = func(ctx context.Context, limit int, maxAttempts int, staleBefore time.Time) ([]PendingExport, error) {
return []PendingExport{
{ExportID: exportID, Row: Row{Question: "q", Answer: "a", AskerName: "x", AskerUnionID: "union-x"}, Attempts: 0},
}, nil
}
d, err := NewDispatcher(repo, inserter, nil)
if err != nil {
t.Fatalf("NewDispatcher: %v", err)
}
stats, err := d.RunOnce(context.Background())
if err != nil {
t.Fatalf("RunOnce: %v", err)
}
if stats.Claimed != 1 {
t.Errorf("Claimed = %d, want 1", stats.Claimed)
}
if stats.Exported != 0 {
t.Errorf("Exported = %d, want 0", stats.Exported)
}
if stats.Failed != 1 {
t.Errorf("Failed = %d, want 1", stats.Failed)
}
// Check failure code is stable, not raw error.
repo.mu.Lock()
defer repo.mu.Unlock()
if len(repo.failed) != 1 {
t.Fatalf("failed records = %d, want 1", len(repo.failed))
}
if repo.failed[0].ExportID != exportID {
t.Errorf("failed exportID = %q, want %q", repo.failed[0].ExportID, exportID)
}
if repo.failed[0].FailureCode != FailureCodeRemoteInsertFailed {
t.Errorf("failure code = %q, want %q", repo.failed[0].FailureCode, FailureCodeRemoteInsertFailed)
}
// The raw error with secrets must not appear in stats or returned values.
// (Already verified: stats don't leak error strings.)
}
// ---------------------------------------------------------------------------
// Test 4: blocking inserter respects per-item timeout
// ---------------------------------------------------------------------------
func TestDispatcherRunOncePerItemTimeout(t *testing.T) {
repo := newStubRepo()
inserter := newStubInserter()
inserter.blockCh = make(chan struct{}) // never unblocks
exportID := "exp-timeout"
repo.claimFn = func(ctx context.Context, limit int, maxAttempts int, staleBefore time.Time) ([]PendingExport, error) {
return []PendingExport{
{ExportID: exportID, Row: Row{Question: "q", Answer: "a", AskerName: "x", AskerUnionID: "union-x"}, Attempts: 0},
}, nil
}
opts := &DispatcherOptions{SendTimeout: 50 * time.Millisecond}
d, err := NewDispatcher(repo, inserter, opts)
if err != nil {
t.Fatalf("NewDispatcher: %v", err)
}
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
stats, err := d.RunOnce(ctx)
if err != nil {
t.Fatalf("RunOnce: %v", err)
}
if stats.Failed != 1 {
t.Errorf("Failed = %d, want 1 (timeout should cause mark failed)", stats.Failed)
}
if stats.Exported != 0 {
t.Errorf("Exported = %d, want 0", stats.Exported)
}
// Verify failure code is stable.
repo.mu.Lock()
defer repo.mu.Unlock()
if len(repo.failed) != 1 {
t.Fatalf("failed records = %d, want 1", len(repo.failed))
}
if repo.failed[0].FailureCode != FailureCodeRemoteInsertFailed {
t.Errorf("failure code = %q, want %q", repo.failed[0].FailureCode, FailureCodeRemoteInsertFailed)
}
}
// ---------------------------------------------------------------------------
// Test 5: claim receives batch/max/stale time near expected
// ---------------------------------------------------------------------------
func TestDispatcherRunOnceClaimParams(t *testing.T) {
repo := newStubRepo()
inserter := newStubInserter()
opts := &DispatcherOptions{
Batch: 7,
MaxAttempts: 3,
ClaimLease: 2 * time.Minute,
SendTimeout: 5 * time.Second,
}
d, err := NewDispatcher(repo, inserter, opts)
if err != nil {
t.Fatalf("NewDispatcher: %v", err)
}
before := time.Now()
_, err = d.RunOnce(context.Background())
if err != nil {
t.Fatalf("RunOnce: %v", err)
}
after := time.Now()
repo.mu.Lock()
limit := repo.lastClaimLimit
maxAtt := repo.lastClaimMaxAttempts
staleBefore := repo.lastClaimStaleBefore
repo.mu.Unlock()
if limit != 7 {
t.Errorf("claim limit = %d, want 7", limit)
}
if maxAtt != 3 {
t.Errorf("claim maxAttempts = %d, want 3", maxAtt)
}
// staleBefore should be approximately now - 2m.
expectedStale := before.Add(-2 * time.Minute)
// Allow some tolerance.
if staleBefore.Before(expectedStale.Add(-2*time.Second)) || staleBefore.After(after.Add(-2*time.Minute).Add(2*time.Second)) {
t.Errorf("staleBefore = %v, expected near %v (before=%v, after=%v)", staleBefore, expectedStale, before, after)
}
}
// ---------------------------------------------------------------------------
// Test 6: parent cancellation before claim and between items
// ---------------------------------------------------------------------------
func TestDispatcherRunOnceParentCancelledBeforeClaim(t *testing.T) {
repo := newStubRepo()
inserter := newStubInserter()
d, err := NewDispatcher(repo, inserter, nil)
if err != nil {
t.Fatalf("NewDispatcher: %v", err)
}
ctx, cancel := context.WithCancel(context.Background())
cancel()
_, err = d.RunOnce(ctx)
if err == nil {
t.Fatal("expected error for cancelled context")
}
if !strings.Contains(err.Error(), "context invalid") {
t.Errorf("error = %v, want 'context invalid'", err)
}
}
func TestDispatcherRunOnceParentCancelledBetweenItems(t *testing.T) {
repo := newStubRepo()
inserter := newStubInserter()
inserter.recordID = "rec-1"
ctx, cancel := context.WithCancel(context.Background())
repo.claimFn = func(ctx context.Context, limit int, maxAttempts int, staleBefore time.Time) ([]PendingExport, error) {
return []PendingExport{
{ExportID: "exp-1", Row: Row{Question: "q1", Answer: "a1", AskerName: "x", AskerUnionID: "union-x"}, Attempts: 0},
{ExportID: "exp-2", Row: Row{Question: "q2", Answer: "a2", AskerName: "y", AskerUnionID: "union-y"}, Attempts: 0},
}, nil
}
// Wrap inserter to cancel parent context after the first InsertRecord call.
cancellingInserter := &cancellingInserter{
inner: inserter,
cancel: cancel,
}
d, err := NewDispatcher(repo, cancellingInserter, nil)
if err != nil {
t.Fatalf("NewDispatcher: %v", err)
}
stats, err := d.RunOnce(ctx)
if err == nil {
t.Fatal("expected error for cancelled context between items")
}
if !strings.Contains(err.Error(), "context cancelled during dispatch") {
t.Errorf("error = %v, want 'context cancelled during dispatch'", err)
}
// First item should have been exported before cancellation.
if stats.Exported != 1 {
t.Errorf("Exported = %d, want 1", stats.Exported)
}
}
// cancellingInserter wraps a RecordInserter and cancels a context after the first call.
type cancellingInserter struct {
inner RecordInserter
cancel context.CancelFunc
mu sync.Mutex
calls int
}
func (a *cancellingInserter) InsertRecord(ctx context.Context, row Row) (string, error) {
a.mu.Lock()
a.calls++
calls := a.calls
a.mu.Unlock()
result, err := a.inner.InsertRecord(ctx, row)
if calls == 1 {
a.cancel()
}
return result, err
}
// ---------------------------------------------------------------------------
// Test 7: claim/mark errors and stats semantics
// ---------------------------------------------------------------------------
func TestDispatcherRunOnceClaimError(t *testing.T) {
repo := newStubRepo()
inserter := newStubInserter()
claimErr := errors.New("database connection refused")
repo.claimFn = func(ctx context.Context, limit int, maxAttempts int, staleBefore time.Time) ([]PendingExport, error) {
return nil, claimErr
}
d, err := NewDispatcher(repo, inserter, nil)
if err != nil {
t.Fatalf("NewDispatcher: %v", err)
}
stats, err := d.RunOnce(context.Background())
if err == nil {
t.Fatal("expected error for claim failure")
}
if !strings.Contains(err.Error(), "claim failed") {
t.Errorf("error = %v, want 'claim failed' stage wrapper", err)
}
if stats.Claimed != 0 {
t.Errorf("Claimed = %d, want 0 on claim error", stats.Claimed)
}
}
func TestDispatcherRunOnceMarkExportedError(t *testing.T) {
repo := newStubRepo()
inserter := newStubInserter()
inserter.recordID = "rec-mark-err"
repo.markExportedErr = errors.New("mark exported db error")
exportID := "exp-mark-err"
repo.claimFn = func(ctx context.Context, limit int, maxAttempts int, staleBefore time.Time) ([]PendingExport, error) {
return []PendingExport{
{ExportID: exportID, Row: Row{Question: "q", Answer: "a", AskerName: "x", AskerUnionID: "union-x"}, Attempts: 0},
}, nil
}
d, err := NewDispatcher(repo, inserter, nil)
if err != nil {
t.Fatalf("NewDispatcher: %v", err)
}
stats, err := d.RunOnce(context.Background())
if err == nil {
t.Fatal("expected error for mark exported failure")
}
if !strings.Contains(err.Error(), "mark exported failed") {
t.Errorf("error = %v, want 'mark exported failed' stage wrapper", err)
}
// Stats should reflect the insert succeeded before mark failed.
if stats.Exported != 0 {
t.Errorf("Exported = %d, want 0 (mark failed before increment)", stats.Exported)
}
}
func TestDispatcherRunOnceStatsSemantics(t *testing.T) {
repo := newStubRepo()
inserter := newStubInserter()
inserter.recordID = "rec-ok"
// 2 success, 1 failure.
inserter.perCallErrors = map[int]error{
2: errors.New("network error"),
}
repo.claimFn = func(ctx context.Context, limit int, maxAttempts int, staleBefore time.Time) ([]PendingExport, error) {
return []PendingExport{
{ExportID: "exp-ok1", Row: Row{Question: "q1", Answer: "a1", AskerName: "x", AskerUnionID: "union-x"}, Attempts: 0},
{ExportID: "exp-fail", Row: Row{Question: "q2", Answer: "a2", AskerName: "y", AskerUnionID: "union-y"}, Attempts: 0},
{ExportID: "exp-ok2", Row: Row{Question: "q3", Answer: "a3", AskerName: "z", AskerUnionID: "union-z"}, Attempts: 0},
}, nil
}
d, err := NewDispatcher(repo, inserter, nil)
if err != nil {
t.Fatalf("NewDispatcher: %v", err)
}
stats, err := d.RunOnce(context.Background())
if err != nil {
t.Fatalf("RunOnce: %v", err)
}
if stats.Claimed != 3 {
t.Errorf("Claimed = %d, want 3", stats.Claimed)
}
if stats.Exported != 2 {
t.Errorf("Exported = %d, want 2", stats.Exported)
}
if stats.Failed != 1 {
t.Errorf("Failed = %d, want 1", stats.Failed)
}
}
// ---------------------------------------------------------------------------
// Test 8: constructor invalid bounds, nil, typed-nil
// ---------------------------------------------------------------------------
func TestNewDispatcherNilDeps(t *testing.T) {
inserter := newStubInserter()
_, err := NewDispatcher(nil, inserter, nil)
if err == nil {
t.Fatal("expected error for nil repo")
}
if !strings.Contains(err.Error(), "repo is nil") {
t.Errorf("error = %v, want 'repo is nil'", err)
}
repo := newStubRepo()
_, err = NewDispatcher(repo, nil, nil)
if err == nil {
t.Fatal("expected error for nil inserter")
}
if !strings.Contains(err.Error(), "inserter is nil") {
t.Errorf("error = %v, want 'inserter is nil'", err)
}
}
func TestNewDispatcherTypedNil(t *testing.T) {
// Typed nil: interface value is non-nil but underlying concrete is nil.
var typedNilRepo *stubRepo = nil
var repo ExportRepository = typedNilRepo
inserter := newStubInserter()
_, err := NewDispatcher(repo, inserter, nil)
if err == nil {
t.Fatal("expected error for typed-nil repo")
}
if !strings.Contains(err.Error(), "repo is nil") {
t.Errorf("error = %v, want 'repo is nil'", err)
}
repo2 := newStubRepo()
var typedNilInserter *stubInserter = nil
var inserter2 RecordInserter = typedNilInserter
_, err = NewDispatcher(repo2, inserter2, nil)
if err == nil {
t.Fatal("expected error for typed-nil inserter")
}
if !strings.Contains(err.Error(), "inserter is nil") {
t.Errorf("error = %v, want 'inserter is nil'", err)
}
}
func TestNewDispatcherInvalidBounds(t *testing.T) {
repo := newStubRepo()
inserter := newStubInserter()
tests := []struct {
name string
opts DispatcherOptions
want string
}{
{"negative batch", DispatcherOptions{Batch: -1}, "batch must be >= 0"},
{"negative maxAttempts", DispatcherOptions{MaxAttempts: -1}, "maxAttempts must be >= 0"},
{"negative claimLease", DispatcherOptions{ClaimLease: -1 * time.Second}, "claimLease must be >= 0"},
{"negative sendTimeout", DispatcherOptions{SendTimeout: -1 * time.Second}, "sendTimeout must be >= 0"},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
_, err := NewDispatcher(repo, inserter, &tt.opts)
if err == nil {
t.Fatal("expected error")
}
if !strings.Contains(err.Error(), tt.want) {
t.Errorf("error = %v, want containing %q", err, tt.want)
}
})
}
}
func TestNewDispatcherDefaults(t *testing.T) {
repo := newStubRepo()
inserter := newStubInserter()
d, err := NewDispatcher(repo, inserter, nil)
if err != nil {
t.Fatalf("NewDispatcher: %v", err)
}
if d.batch != 10 {
t.Errorf("batch = %d, want 10", d.batch)
}
if d.maxAttempts != 5 {
t.Errorf("maxAttempts = %d, want 5", d.maxAttempts)
}
if d.claimLease != 5*time.Minute {
t.Errorf("claimLease = %v, want 5m", d.claimLease)
}
if d.sendTimeout != 10*time.Second {
t.Errorf("sendTimeout = %v, want 10s", d.sendTimeout)
}
}
func TestNewDispatcherZeroBatch(t *testing.T) {
// Zero batch is allowed (validates as >= 0, but not > 0 so use default).
repo := newStubRepo()
inserter := newStubInserter()
opts := &DispatcherOptions{Batch: 0}
d, err := NewDispatcher(repo, inserter, opts)
if err != nil {
t.Fatalf("NewDispatcher with batch=0: %v", err)
}
if d.batch != 10 {
t.Errorf("batch = %d, want 10 (default when opts.Batch=0)", d.batch)
}
}
// ---------------------------------------------------------------------------
// Test 9: exact Row passed to inserter
// ---------------------------------------------------------------------------
func TestDispatcherRunOnceExactRowPassed(t *testing.T) {
repo := newStubRepo()
inserter := newStubInserter()
row := Row{
Question: "怎么修?",
Answer: "按步骤操作",
AskedTime: time.Date(2025, 6, 15, 14, 30, 0, 0, time.FixedZone("CST", 8*60*60)),
AskerName: "张三",
AskerUnionID: "union-zhang-san",
TriggerPerson: "李四",
Department: "研发部",
Region: "华东",
}
exportID := "exp-cells"
repo.claimFn = func(ctx context.Context, limit int, maxAttempts int, staleBefore time.Time) ([]PendingExport, error) {
return []PendingExport{
{ExportID: exportID, Row: row, Attempts: 0},
}, nil
}
d, err := NewDispatcher(repo, inserter, nil)
if err != nil {
t.Fatalf("NewDispatcher: %v", err)
}
_, err = d.RunOnce(context.Background())
if err != nil {
t.Fatalf("RunOnce: %v", err)
}
inserter.mu.Lock()
defer inserter.mu.Unlock()
if len(inserter.rows) != 1 {
t.Fatalf("inserter received %d rows, want 1", len(inserter.rows))
}
got := inserter.rows[0]
if got.Question != row.Question {
t.Errorf("Question = %q, want %q", got.Question, row.Question)
}
if got.Answer != row.Answer {
t.Errorf("Answer = %q, want %q", got.Answer, row.Answer)
}
if !got.AskedTime.Equal(row.AskedTime) {
t.Errorf("AskedTime = %v, want %v", got.AskedTime, row.AskedTime)
}
if got.AskerName != row.AskerName {
t.Errorf("AskerName = %q, want %q", got.AskerName, row.AskerName)
}
if got.AskerUnionID != row.AskerUnionID {
t.Errorf("AskerUnionID = %q, want %q", got.AskerUnionID, row.AskerUnionID)
}
if got.TriggerPerson != row.TriggerPerson {
t.Errorf("TriggerPerson = %q, want %q", got.TriggerPerson, row.TriggerPerson)
}
if got.Department != row.Department {
t.Errorf("Department = %q, want %q", got.Department, row.Department)
}
if got.Region != row.Region {
t.Errorf("Region = %q, want %q", got.Region, row.Region)
}
}
// ---------------------------------------------------------------------------
// Additional: RunOnce returns error when sendCtx expires but parent OK
// ---------------------------------------------------------------------------
func TestDispatcherRunOnceSendTimeoutDoesNotKillParent(t *testing.T) {
repo := newStubRepo()
inserter := newStubInserter()
inserter.blockCh = make(chan struct{}) // never unblocks
exportID := "exp-timeout2"
repo.claimFn = func(ctx context.Context, limit int, maxAttempts int, staleBefore time.Time) ([]PendingExport, error) {
return []PendingExport{
{ExportID: exportID, Row: Row{Question: "q", Answer: "a", AskerName: "x", AskerUnionID: "union-x"}, Attempts: 0},
}, nil
}
opts := &DispatcherOptions{SendTimeout: 20 * time.Millisecond}
d, err := NewDispatcher(repo, inserter, opts)
if err != nil {
t.Fatalf("NewDispatcher: %v", err)
}
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
stats, err := d.RunOnce(ctx)
if err != nil {
t.Fatalf("RunOnce should not fail on per-item timeout: %v", err)
}
if stats.Failed != 1 {
t.Errorf("Failed = %d, want 1", stats.Failed)
}
// Parent context should still be alive.
if ctx.Err() != nil {
t.Errorf("parent context should still be alive, got %v", ctx.Err())
}
}
// ---------------------------------------------------------------------------
// Additional: RecordInserter interface satisfaction check
// ---------------------------------------------------------------------------
func TestClientSatisfiesRecordInserter(t *testing.T) {
// Compile-time check: *Client implements RecordInserter.
var _ RecordInserter = (*Client)(nil)
}
// ---------------------------------------------------------------------------
// Error leakage: repository errors must not leak secrets
// ---------------------------------------------------------------------------
func TestDispatcherRunOnceClaimErrorNoLeak(t *testing.T) {
repo := newStubRepo()
inserter := newStubInserter()
// Claim error containing SQL/turn/export/person secrets.
repo.claimFn = func(ctx context.Context, limit int, maxAttempts int, staleBefore time.Time) ([]PendingExport, error) {
return nil, fmt.Errorf("SQLSTATE 42P01: relation \"turn_export_person\" does not exist")
}
d, err := NewDispatcher(repo, inserter, nil)
if err != nil {
t.Fatalf("NewDispatcher: %v", err)
}
_, err = d.RunOnce(context.Background())
if err == nil {
t.Fatal("expected error for claim failure")
}
errStr := err.Error()
secrets := []string{"SQLSTATE", "42P01", "turn_export_person", "relation"}
for _, s := range secrets {
if strings.Contains(errStr, s) {
t.Errorf("error leaks secret %q: %v", s, errStr)
}
}
}
func TestDispatcherRunOnceMarkFailedNoLeak(t *testing.T) {
repo := newStubRepo()
inserter := newStubInserter()
inserter.err = errors.New("some insert error")
repo.markFailedErr = fmt.Errorf("SQLSTATE 23505: duplicate key in export_person_report")
exportID := "exp-leak-mf"
repo.claimFn = func(ctx context.Context, limit int, maxAttempts int, staleBefore time.Time) ([]PendingExport, error) {
return []PendingExport{
{ExportID: exportID, Row: Row{Question: "q", Answer: "a", AskerName: "x", AskerUnionID: "union-x"}, Attempts: 0},
}, nil
}
d, err := NewDispatcher(repo, inserter, nil)
if err != nil {
t.Fatalf("NewDispatcher: %v", err)
}
_, err = d.RunOnce(context.Background())
if err == nil {
t.Fatal("expected error for mark failed failure")
}
errStr := err.Error()
secrets := []string{"SQLSTATE", "23505", "export_person_report", "duplicate"}
for _, s := range secrets {
if strings.Contains(errStr, s) {
t.Errorf("error leaks secret %q: %v", s, errStr)
}
}
}
func TestDispatcherRunOnceMarkExportedNoLeak(t *testing.T) {
repo := newStubRepo()
inserter := newStubInserter()
inserter.recordID = "rec-leak"
repo.markExportedErr = fmt.Errorf("SQLSTATE 08006: connection lost for turn_export_write")
exportID := "exp-leak-me"
repo.claimFn = func(ctx context.Context, limit int, maxAttempts int, staleBefore time.Time) ([]PendingExport, error) {
return []PendingExport{
{ExportID: exportID, Row: Row{Question: "q", Answer: "a", AskerName: "x", AskerUnionID: "union-x"}, Attempts: 0},
}, nil
}
d, err := NewDispatcher(repo, inserter, nil)
if err != nil {
t.Fatalf("NewDispatcher: %v", err)
}
_, err = d.RunOnce(context.Background())
if err == nil {
t.Fatal("expected error for mark exported failure")
}
errStr := err.Error()
secrets := []string{"SQLSTATE", "08006", "turn_export_write", "connection"}
for _, s := range secrets {
if strings.Contains(errStr, s) {
t.Errorf("error leaks secret %q: %v", s, errStr)
}
}
}
// ---------------------------------------------------------------------------
// Value-type constructor: non-pointer implementations should not panic
// ---------------------------------------------------------------------------
// valueRepo is a non-pointer ExportRepository implementation.
type valueRepo struct{}
func (v valueRepo) Claim(ctx context.Context, limit int, maxAttempts int, staleBefore time.Time) ([]PendingExport, error) {
return nil, nil
}
func (v valueRepo) MarkExported(ctx context.Context, exportID, location string) error { return nil }
func (v valueRepo) MarkFailed(ctx context.Context, exportID, failureCode string) error { return nil }
// valueInserter is a non-pointer RecordInserter implementation.
type valueInserter struct{}
func (v valueInserter) InsertRecord(ctx context.Context, row Row) (string, error) {
return "rec-value-001", nil
}
func TestNewDispatcherValueTypeRepo(t *testing.T) {
// value-type repo should not panic (was: reflect.ValueOf(x).IsNil() on non-nilable kind).
var repo ExportRepository = valueRepo{}
inserter := newStubInserter()
d, err := NewDispatcher(repo, inserter, nil)
if err != nil {
t.Fatalf("NewDispatcher with value-type repo: %v", err)
}
_ = d
}
func TestNewDispatcherValueTypeInserter(t *testing.T) {
repo := newStubRepo()
var inserter RecordInserter = valueInserter{}
d, err := NewDispatcher(repo, inserter, nil)
if err != nil {
t.Fatalf("NewDispatcher with value-type inserter: %v", err)
}
_ = d
}
func TestNewDispatcherValueTypeBoth(t *testing.T) {
var repo ExportRepository = valueRepo{}
var inserter RecordInserter = valueInserter{}
d, err := NewDispatcher(repo, inserter, nil)
if err != nil {
t.Fatalf("NewDispatcher with value-type both: %v", err)
}
_ = d
}
+136
View File
@@ -0,0 +1,136 @@
package questiontable
import (
"strings"
"time"
"unicode"
"unicode/utf8"
)
// Field IDs for the DingTalk Notable v1 question table.
// Verified against the live Notable v1 base.
const (
FieldChineseQuestion = "ZaMZF6j" // text
FieldAskedTime = "SCug9My" // date (Unix milliseconds)
FieldAsker = "S2XA1uZ" // user (array of {name, unionId})
FieldOriginalQuestion = "bETcL3r" // text
FieldAnswer = "Y4cpmxU" // text
FieldTriggerPerson = "JgY2f1z" // text
FieldLanguage = "DNIV3mQ" // text
FieldDepartment = "eBc6a6b" // text
FieldRegion = "fPNLeVW" // text
)
// Row holds the data needed to produce a Notable v1 typed record.
type Row struct {
Question string
Answer string
AskedTime time.Time
AskerName string
AskerUnionID string
TriggerPerson string
Department string
Region string
}
// ToFields returns a typed field-ID → value map suitable for the Notable v1
// InsertRecord request body. The caller is responsible for the semantics of
// each field; this method only handles mapping, language detection, time
// formatting, and formula-injection sanitisation.
// The five v1 select fields (QMn9UoU, H3gtTuv, nFvv8OR, UMqUPAG, lnF7gF9)
// are omitted entirely.
func (r Row) ToFields() map[string]interface{} {
fields := make(map[string]interface{})
lang := detectLanguage(r.Question)
// ChineseQuestion: only when Han present.
if lang == "中文" {
fields[FieldChineseQuestion] = sanitise(r.Question)
}
// Answer.
fields[FieldAnswer] = sanitise(r.Answer)
// AskedTime: Unix milliseconds if non-zero; omit if zero.
if !r.AskedTime.IsZero() {
fields[FieldAskedTime] = r.AskedTime.UnixMilli()
}
// Asker: user object only when AskerUnionID is non-blank.
if strings.TrimSpace(r.AskerUnionID) != "" {
fields[FieldAsker] = []interface{}{
map[string]interface{}{
"name": strings.TrimSpace(r.AskerName),
"unionId": strings.TrimSpace(r.AskerUnionID),
},
}
}
// OriginalQuestion.
fields[FieldOriginalQuestion] = sanitise(r.Question)
// Language.
fields[FieldLanguage] = sanitise(lang)
// TriggerPerson: fallback to AskerName.
triggerPerson := r.TriggerPerson
if triggerPerson == "" {
triggerPerson = r.AskerName
}
fields[FieldTriggerPerson] = sanitise(triggerPerson)
// Department.
fields[FieldDepartment] = sanitise(r.Department)
// Region.
fields[FieldRegion] = sanitise(r.Region)
return fields
}
// detectLanguage returns "中文" when s contains any Han character,
// "英文" when it contains Latin letters but no Han,
// and "其他" otherwise.
func detectLanguage(s string) string {
hasHan := false
hasLatin := false
for _, r := range s {
if isHan(r) {
hasHan = true
}
if isLatin(r) {
hasLatin = true
}
}
if hasHan {
return "中文"
}
if hasLatin {
return "英文"
}
return "其他"
}
func isHan(r rune) bool {
return unicode.Is(unicode.Han, r)
}
func isLatin(r rune) bool {
return unicode.Is(unicode.Latin, r)
}
// sanitise protects against spreadsheet formula injection. If the string
// begins with =, +, -, or @, a single apostrophe is prefixed. The string
// is otherwise returned unchanged.
func sanitise(s string) string {
if s == "" {
return s
}
r, _ := utf8.DecodeRuneInString(s)
switch r {
case '=', '+', '-', '@':
return "'" + s
}
return s
}
+356
View File
@@ -0,0 +1,356 @@
package questiontable
import (
"testing"
"time"
)
func TestRowToFieldsExactMapping(t *testing.T) {
shanghai, _ := time.LoadLocation("Asia/Shanghai")
asked := time.Date(2025, 6, 15, 14, 30, 0, 0, shanghai)
r := Row{
Question: "这个怎么修?",
Answer: "请按以下步骤操作。",
AskedTime: asked,
AskerName: "张三",
AskerUnionID: "union-zhang-san",
TriggerPerson: "李四",
Department: "研发部",
Region: "华东",
}
fields := r.ToFields()
// ChineseQuestion (only when Han present).
if fields[FieldChineseQuestion] != "这个怎么修?" {
t.Errorf("ChineseQuestion = %q, want %q", fields[FieldChineseQuestion], "这个怎么修?")
}
// Answer.
if fields[FieldAnswer] != "请按以下步骤操作。" {
t.Errorf("Answer = %q, want %q", fields[FieldAnswer], "请按以下步骤操作。")
}
// AskedTime: Unix milliseconds.
expectedMS := asked.UnixMilli()
if fields[FieldAskedTime] != expectedMS {
t.Errorf("AskedTime = %v (%T), want %v (int64)", fields[FieldAskedTime], fields[FieldAskedTime], expectedMS)
}
// Asker: user array.
userArr, ok := fields[FieldAsker].([]interface{})
if !ok {
t.Fatalf("Asker not a []interface{}, got %T", fields[FieldAsker])
}
if len(userArr) != 1 {
t.Fatalf("Asker len = %d, want 1", len(userArr))
}
userObj, ok := userArr[0].(map[string]interface{})
if !ok {
t.Fatalf("Asker[0] not a map[string]interface{}, got %T", userArr[0])
}
if userObj["name"] != "张三" {
t.Errorf("Asker name = %q, want %q", userObj["name"], "张三")
}
if userObj["unionId"] != "union-zhang-san" {
t.Errorf("Asker unionId = %q, want %q", userObj["unionId"], "union-zhang-san")
}
// OriginalQuestion.
if fields[FieldOriginalQuestion] != "这个怎么修?" {
t.Errorf("OriginalQuestion = %q, want %q", fields[FieldOriginalQuestion], "这个怎么修?")
}
// Language.
if fields[FieldLanguage] != "中文" {
t.Errorf("Language = %q, want %q", fields[FieldLanguage], "中文")
}
// TriggerPerson.
if fields[FieldTriggerPerson] != "李四" {
t.Errorf("TriggerPerson = %q, want %q", fields[FieldTriggerPerson], "李四")
}
// Department.
if fields[FieldDepartment] != "研发部" {
t.Errorf("Department = %q, want %q", fields[FieldDepartment], "研发部")
}
// Region.
if fields[FieldRegion] != "华东" {
t.Errorf("Region = %q, want %q", fields[FieldRegion], "华东")
}
// Omitted: the five select fields must not be present.
omitted := []string{"QMn9UoU", "H3gtTuv", "nFvv8OR", "UMqUPAG", "lnF7gF9"}
for _, fid := range omitted {
if _, exists := fields[fid]; exists {
t.Errorf("select field %q should be omitted but is present", fid)
}
}
}
func TestRowToFieldsDateZeroOmission(t *testing.T) {
r := Row{
Question: "test",
Answer: "answer",
AskerName: "user",
AskerUnionID: "union-1",
// AskedTime is zero.
}
fields := r.ToFields()
if _, exists := fields[FieldAskedTime]; exists {
t.Error("AskedTime field should be omitted when zero")
}
}
func TestRowToFieldsDateInt64UnixMilli(t *testing.T) {
shanghai, _ := time.LoadLocation("Asia/Shanghai")
asked := time.Date(2026, 1, 2, 3, 4, 5, 0, shanghai)
r := Row{
Question: "test",
Answer: "answer",
AskedTime: asked,
AskerName: "user",
AskerUnionID: "union-1",
}
fields := r.ToFields()
got, ok := fields[FieldAskedTime].(int64)
if !ok {
t.Fatalf("AskedTime is %T, want int64", fields[FieldAskedTime])
}
if got != asked.UnixMilli() {
t.Errorf("AskedTime = %d, want %d", got, asked.UnixMilli())
}
}
func TestRowToFieldsUserShape(t *testing.T) {
// With union ID: user array present.
r := Row{
Question: "test",
Answer: "answer",
AskerName: " 张三 ",
AskerUnionID: " union-123 ",
}
fields := r.ToFields()
userArr, ok := fields[FieldAsker].([]interface{})
if !ok {
t.Fatalf("Asker not present or wrong type: %T", fields[FieldAsker])
}
userObj := userArr[0].(map[string]interface{})
if userObj["name"] != "张三" {
t.Errorf("name = %q, want %q (trimmed)", userObj["name"], "张三")
}
if userObj["unionId"] != "union-123" {
t.Errorf("unionId = %q, want %q (trimmed)", userObj["unionId"], "union-123")
}
// Without union ID: Asker field omitted.
r2 := Row{
Question: "test",
Answer: "answer",
AskerName: "张三",
AskerUnionID: "",
}
fields2 := r2.ToFields()
if _, exists := fields2[FieldAsker]; exists {
t.Error("Asker field should be omitted when union ID is blank")
}
// Whitespace-only union ID: omitted.
r3 := Row{
Question: "test",
Answer: "answer",
AskerName: "张三",
AskerUnionID: " ",
}
fields3 := r3.ToFields()
if _, exists := fields3[FieldAsker]; exists {
t.Error("Asker field should be omitted when union ID is whitespace-only")
}
}
func TestRowToFieldsChineseQuestionPolicy(t *testing.T) {
// English question: ChineseQuestion omitted.
r := Row{
Question: "How to fix this?",
Answer: "Follow these steps.",
AskerName: "Alice",
AskerUnionID: "union-alice",
}
fields := r.ToFields()
if _, exists := fields[FieldChineseQuestion]; exists {
t.Error("ChineseQuestion should be omitted for English question")
}
if fields[FieldLanguage] != "英文" {
t.Errorf("Language = %q, want 英文", fields[FieldLanguage])
}
// Chinese question: ChineseQuestion contains the question.
r2 := Row{
Question: "怎么解决?",
Answer: "按步骤操作。",
AskerName: "Bob",
AskerUnionID: "union-bob",
}
fields2 := r2.ToFields()
if fields2[FieldChineseQuestion] != "怎么解决?" {
t.Errorf("ChineseQuestion = %q, want %q", fields2[FieldChineseQuestion], "怎么解决?")
}
if fields2[FieldLanguage] != "中文" {
t.Errorf("Language = %q, want 中文", fields2[FieldLanguage])
}
}
func TestDetectLanguage(t *testing.T) {
tests := []struct {
input string
want string
}{
{"你好", "中文"},
{"Hello", "英文"},
{"Hello世界", "中文"}, // Han takes priority
{"12345", "其他"},
{"", "其他"},
{"こんにちは", "其他"}, // Japanese kana
{"Привет", "其他"}, // Cyrillic
{"Español", "英文"}, // accented Latin via unicode.Latin
{"café", "英文"}, // accented Latin
{"Déjà vu", "英文"}, // accented Latin
}
for _, tt := range tests {
got := detectLanguage(tt.input)
if got != tt.want {
t.Errorf("detectLanguage(%q) = %q, want %q", tt.input, got, tt.want)
}
}
}
func TestSanitiseFormulaInjection(t *testing.T) {
tests := []struct {
input string
want string
}{
{"=SUM(A1:A10)", "'=SUM(A1:A10)"},
{"+A1", "'+A1"},
{"-A1", "'-A1"},
{"@SUM", "'@SUM"},
{"normal text", "normal text"},
{"=", "'="}, // = is the first rune, prefixed to '='
{"", ""},
{" leading space", " leading space"},
{"1=2", "1=2"}, // doesn't start with a formula char
}
for _, tt := range tests {
got := sanitise(tt.input)
if got != tt.want {
t.Errorf("sanitise(%q) = %q, want %q", tt.input, got, tt.want)
}
}
}
func TestSanitiseDoesNotMutateNonFormula(t *testing.T) {
inputs := []string{"hello", "世界", "123", "=SUM(", " a", "a=", "null"}
for _, in := range inputs {
got := sanitise(in)
// Only "=SUM(" should be prefixed.
switch in {
case "=SUM(":
if got != "'=SUM(" {
t.Errorf("sanitise(%q) = %q, want '=SUM(", in, got)
}
default:
if got != in {
t.Errorf("sanitise(%q) = %q, want %q (no change)", in, got, in)
}
}
}
}
func TestTriggerPersonFallback(t *testing.T) {
// When TriggerPerson is empty, fall back to AskerName.
r := Row{
Question: "test",
Answer: "answer",
AskerName: "张三",
AskerUnionID: "union-1",
TriggerPerson: "",
}
fields := r.ToFields()
if fields[FieldTriggerPerson] != "张三" {
t.Errorf("TriggerPerson = %q, want %q (fallback to asker)", fields[FieldTriggerPerson], "张三")
}
// When TriggerPerson is set, use it.
r2 := Row{
Question: "test",
Answer: "answer",
AskerName: "张三",
AskerUnionID: "union-1",
TriggerPerson: "nickname",
}
fields2 := r2.ToFields()
if fields2[FieldTriggerPerson] != "nickname" {
t.Errorf("TriggerPerson = %q, want %q (use trigger)", fields2[FieldTriggerPerson], "nickname")
}
}
func TestFormulaInjectionOnAllColumns(t *testing.T) {
shanghai, _ := time.LoadLocation("Asia/Shanghai")
asked := time.Date(2025, 1, 1, 0, 0, 0, 0, shanghai)
// Use a question that starts with formula char AND contains Han so
// the Chinese-question field is populated.
r := Row{
Question: "=修复()",
Answer: "+cmd",
AskedTime: asked,
AskerName: "-张三",
AskerUnionID: "union-sanitise",
TriggerPerson: "@trigger",
Department: "=DEPT",
Region: "normal",
}
fields := r.ToFields()
// Chinese question: "=修复()" has Han, so it's Chinese and gets sanitised.
if fields[FieldChineseQuestion] != "'=修复()" {
t.Errorf("ChineseQuestion = %q, want %q", fields[FieldChineseQuestion], "'=修复()")
}
// Answer: "+cmd" → "'+cmd"
if fields[FieldAnswer] != "'+cmd" {
t.Errorf("Answer = %q, want %q", fields[FieldAnswer], "'+cmd")
}
// OriginalQuestion: "=修复()" → "'=修复()"
if fields[FieldOriginalQuestion] != "'=修复()" {
t.Errorf("OriginalQuestion = %q, want %q", fields[FieldOriginalQuestion], "'=修复()")
}
// TriggerPerson: "@trigger" → "'@trigger"
if fields[FieldTriggerPerson] != "'@trigger" {
t.Errorf("TriggerPerson = %q, want %q", fields[FieldTriggerPerson], "'@trigger")
}
// Department: "=DEPT" → "'=DEPT"
if fields[FieldDepartment] != "'=DEPT" {
t.Errorf("Department = %q, want %q", fields[FieldDepartment], "'=DEPT")
}
// Region: "normal" → no change
if fields[FieldRegion] != "normal" {
t.Errorf("Region = %q, want %q", fields[FieldRegion], "normal")
}
// Asker name: "-张三" → "'-张三" (via user object)
userArr := fields[FieldAsker].([]interface{})
userObj := userArr[0].(map[string]interface{})
// Asker name is NOT sanitised per the contract (only text fields are sanitised),
// but the name is trimmed. Let's verify — the task says trim only union ID/name,
// preserve question/answer text. So name is sanitised via TrimSpace, not sanitise().
// Actually, sanitise is only applied to the text fields. Asker name goes through
// strings.TrimSpace but not sanitise. Let's check the behavior.
// The task says: "Trim only union ID/name for the user object if existing
// conventions require; preserve question/answer text." So name is trimmed, not
// sanitised. We should verify the trimmed name.
if userObj["name"] != "-张三" {
t.Errorf("Asker name = %q, want %q", userObj["name"], "-张三")
}
}
@@ -14,8 +14,10 @@ func TestBeginTurnStoresDirectorySnapshotsAndGroupTitle(t *testing.T) {
src := strings.ToUpper(string(raw))
for _, required := range []string{
"DEPARTMENT_SNAPSHOT,REGION_SNAPSHOT",
"ACTOR_UNION_ID_SNAPSHOT",
"TURN.DEPARTMENTSNAPSHOT",
"TURN.REGIONSNAPSHOT",
"TURN.ACTORUNIONIDSNAPSHOT",
"TURN.CONVERSATIONTITLE",
} {
if !strings.Contains(src, required) {
@@ -0,0 +1,205 @@
// Package postgres contains source-level contract tests for the
// question_table_exports outbox: the store must enqueue inside
// FinishTurn's transaction, guarded by shouldCreateAnswerMessage.
// No database connection is required.
package postgres
import (
"os"
"strings"
"testing"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/core"
)
// TestFinishTurnBeginTurnSnapshot verifies that the BeginTurn INSERT
// into messages includes trigger_person_snapshot, actor_union_id_snapshot,
// department_snapshot, and region_snapshot columns.
func TestFinishTurnBeginTurnSnapshot(t *testing.T) {
data, err := os.ReadFile("store.go")
if err != nil {
t.Fatalf("read store.go: %v", err)
}
src := string(data)
// The BeginTurn message INSERT must include trigger_person_snapshot
if !strings.Contains(src, "trigger_person_snapshot") {
t.Error("BeginTurn message INSERT must include trigger_person_snapshot column")
}
if !strings.Contains(src, "actor_union_id_snapshot") {
t.Error("BeginTurn message INSERT must include actor_union_id_snapshot column")
}
if !strings.Contains(src, "department_snapshot") {
t.Error("BeginTurn message INSERT must include department_snapshot column")
}
if !strings.Contains(src, "region_snapshot") {
t.Error("BeginTurn message INSERT must include region_snapshot column")
}
// The VALUES arguments must include turn.ActorUnionIDSnapshot
if !strings.Contains(src, "turn.ActorUnionIDSnapshot") {
t.Error("BeginTurn message INSERT VALUES must include turn.ActorUnionIDSnapshot argument")
}
}
func TestShouldEnqueueQuestionTableExportOnlyForDingTalkSucceededAnswers(t *testing.T) {
tests := []struct {
name string
turn core.Turn
want bool
}{
{name: "dingtalk success", turn: core.Turn{Channel: "dingtalk", Status: "succeeded", Answer: "answer", AnswerMessageID: "answer-1"}, want: true},
{name: "dingtalk case and space", turn: core.Turn{Channel: " DingTalk ", Status: "succeeded", Answer: "answer", AnswerMessageID: "answer-1"}, want: true},
{name: "web success", turn: core.Turn{Channel: "web", Status: "succeeded", Answer: "answer", AnswerMessageID: "answer-1"}},
{name: "http success", turn: core.Turn{Channel: "http", Status: "succeeded", Answer: "answer", AnswerMessageID: "answer-1"}},
{name: "empty channel", turn: core.Turn{Status: "succeeded", Answer: "answer", AnswerMessageID: "answer-1"}},
{name: "dingtalk failed", turn: core.Turn{Channel: "dingtalk", Status: "failed", Answer: "answer", AnswerMessageID: "answer-1"}},
{name: "dingtalk empty answer", turn: core.Turn{Channel: "dingtalk", Status: "succeeded", Answer: " ", AnswerMessageID: "answer-1"}},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
if got := shouldEnqueueQuestionTableExport(tc.turn); got != tc.want {
t.Fatalf("shouldEnqueueQuestionTableExport()=%v, want %v", got, tc.want)
}
})
}
}
// TestFinishTurnOutboxInsideTransaction verifies that the
// question_table_exports INSERT is inside the FinishTurn transaction
// (between tx begin and tx commit), uses 'pending' status, and has
// ON CONFLICT (tenant_key,channel,turn_id) DO NOTHING.
func TestFinishTurnOutboxInsideTransaction(t *testing.T) {
data, err := os.ReadFile("store.go")
if err != nil {
t.Fatalf("read store.go: %v", err)
}
src := string(data)
// Extract the FinishTurn function body using deterministic boundaries
finishStart := strings.Index(src, "func (s *Store) FinishTurn(")
if finishStart < 0 {
t.Fatal("cannot find FinishTurn signature")
}
finishEnd := strings.Index(src[finishStart:], "func (s *Store) RecordDelivery(")
if finishEnd < 0 {
t.Fatal("cannot find RecordDelivery signature after FinishTurn")
}
body := src[finishStart : finishStart+finishEnd]
// The outbox INSERT must use question_table_exports
if !strings.Contains(body, "question_table_exports") {
t.Error("FinishTurn must INSERT into question_table_exports")
}
// Must use 'pending' status for the outbox
if !strings.Contains(body, "'pending'") {
t.Error("question_table_exports INSERT must use status='pending'")
}
// Must use ON CONFLICT (tenant_key,channel,turn_id) DO NOTHING
if !strings.Contains(body, "ON CONFLICT (tenant_key,channel,turn_id) DO NOTHING") &&
!strings.Contains(body, "ON CONFLICT (tenant_key, channel, turn_id) DO NOTHING") {
t.Error("question_table_exports INSERT must use ON CONFLICT (tenant_key,channel,turn_id) DO NOTHING")
}
// The outbox INSERT must be after the tx begin and before tx commit
txBegin := strings.Index(body, "BeginTx")
txCommit := strings.Index(body, "tx.Commit")
outboxPos := strings.Index(body, "question_table_exports")
if txBegin < 0 || txCommit < 0 || outboxPos < 0 {
t.Fatal("cannot locate transaction boundaries or outbox insert")
}
if outboxPos < txBegin || outboxPos > txCommit {
t.Error("question_table_exports INSERT must be inside the transaction (between BeginTx and Commit)")
}
}
// TestFinishTurnOutboxGuardedByShouldCreateAnswerMessage verifies that
// the outbox INSERT is structurally inside the shouldCreateAnswerMessage
// guard. Failed, processing, and empty-answer turns must not enqueue.
func TestFinishTurnOutboxGuardedByShouldCreateAnswerMessage(t *testing.T) {
data, err := os.ReadFile("store.go")
if err != nil {
t.Fatalf("read store.go: %v", err)
}
src := string(data)
// The shouldCreateAnswerMessage function must guard on status succeeded,
// non-empty answer, and non-empty answer message ID
if !strings.Contains(src, "shouldCreateAnswerMessage") {
t.Fatal("shouldCreateAnswerMessage function must exist")
}
// Extract shouldCreateAnswerMessage function body using deterministic boundaries
shouldFuncStart := strings.Index(src, "func shouldCreateAnswerMessage(")
if shouldFuncStart < 0 {
t.Fatal("cannot find shouldCreateAnswerMessage signature")
}
// Find the closing brace of the function: the first top-level `}` after the signature
// The function body is simple: one return statement, so find the `}` that follows
shouldFuncBodyStart := strings.Index(src[shouldFuncStart:], "{")
if shouldFuncBodyStart < 0 {
t.Fatal("cannot find shouldCreateAnswerMessage opening brace")
}
shouldFuncBodyStart += shouldFuncStart + 1 // position after "{"
shouldFuncEnd := strings.Index(src[shouldFuncBodyStart:], "\n}\n")
if shouldFuncEnd < 0 {
t.Fatal("cannot find shouldCreateAnswerMessage closing brace")
}
shouldFuncEnd += shouldFuncBodyStart + 1 // position of the closing "}"
shouldBody := src[shouldFuncBodyStart:shouldFuncEnd]
if !strings.Contains(shouldBody, "succeeded") {
t.Error("shouldCreateAnswerMessage must check status == succeeded")
}
if !strings.Contains(shouldBody, "turn.Answer") {
t.Error("shouldCreateAnswerMessage must check non-empty answer")
}
if !strings.Contains(shouldBody, "turn.AnswerMessageID") {
t.Error("shouldCreateAnswerMessage must check non-empty AnswerMessageID")
}
// In FinishTurn, extract the guarded block using deterministic boundaries
finishStart := strings.Index(src, "func (s *Store) FinishTurn(")
if finishStart < 0 {
t.Fatal("cannot find FinishTurn signature")
}
finishEnd := strings.Index(src[finishStart:], "func (s *Store) RecordDelivery(")
if finishEnd < 0 {
t.Fatal("cannot find RecordDelivery signature after FinishTurn")
}
finishBody := src[finishStart : finishStart+finishEnd]
// Locate the guarded block: start at "if shouldCreateAnswerMessage(turn) {"
guardStart := strings.Index(finishBody, "if shouldCreateAnswerMessage(turn) {")
if guardStart < 0 {
t.Fatal("cannot find shouldCreateAnswerMessage guard in FinishTurn")
}
// The guarded block ends before the "UPDATE turns SET" statement that follows it
guardEnd := strings.Index(finishBody[guardStart:], "UPDATE turns SET")
if guardEnd < 0 {
t.Fatal("cannot find UPDATE turns SET after guarded block")
}
guardBlock := finishBody[guardStart : guardStart+guardEnd]
if !strings.Contains(guardBlock, "question_table_exports") {
t.Error("question_table_exports INSERT must be inside the shouldCreateAnswerMessage guard block")
}
if !strings.Contains(guardBlock, "if shouldEnqueueQuestionTableExport(turn) {") {
t.Error("question_table_exports INSERT must be guarded to DingTalk turns explicitly")
}
// The guard block must be the one that contains the messages INSERT
// (answer message creation), and after that the outbox INSERT
msgInsert := strings.Index(guardBlock, "INSERT INTO messages")
outboxInsert := strings.Index(guardBlock, "question_table_exports")
if msgInsert < 0 || outboxInsert < 0 {
t.Fatal("cannot locate messages INSERT or outbox INSERT in guard block")
}
if outboxInsert < msgInsert {
t.Error("outbox INSERT should come after answer message INSERT")
}
}
@@ -0,0 +1,219 @@
package postgres
import (
"context"
"errors"
"fmt"
"strings"
"time"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/questiontable"
)
var (
errExportIDRequired = errors.New("questiontable export: exportID is required")
errLocationRequired = errors.New("questiontable export: external record id is required")
errBadExportFailure = errors.New("questiontable export: only remote_question_table_insert_failed is accepted")
errExportRowsAffected = errors.New("questiontable export: RowsAffected != 1")
errExportNilStore = errors.New("questiontable export: store is nil")
)
// QuestionTableExportRepository implements questiontable.ExportRepository
// using a dedicated adapter type. It does NOT add methods to *Store directly
// to avoid conflicts with the existing coverage.NotificationRepository Claim
// and MarkFailed methods.
type QuestionTableExportRepository struct {
store *Store
}
// compile-time interface assertion
var _ questiontable.ExportRepository = (*QuestionTableExportRepository)(nil)
// NewQuestionTableExportRepository creates a validated repository adapter.
// It returns a stable error when store is nil or its pool is nil.
func NewQuestionTableExportRepository(store *Store) (*QuestionTableExportRepository, error) {
if store == nil || store.pool == nil {
return nil, errExportNilStore
}
return &QuestionTableExportRepository{store: store}, nil
}
// Claim implements questiontable.ExportRepository. It atomically claims up to
// limit pending exports whose attempt count is below maxAttempts and whose last
// claim time is before staleBefore. A single CTE with FOR UPDATE SKIP LOCKED
// and one UPDATE is used for atomicity.
func (r *QuestionTableExportRepository) Claim(ctx context.Context, limit int, maxAttempts int, staleBefore time.Time) ([]questiontable.PendingExport, error) {
if r == nil || r.store == nil || r.store.pool == nil {
return nil, errExportNilStore
}
if limit <= 0 || limit > 1000 {
return nil, fmt.Errorf("questiontable export: limit must be 1..1000, got %d", limit)
}
if maxAttempts <= 0 || maxAttempts > 100 {
return nil, fmt.Errorf("questiontable export: maxAttempts must be 1..100, got %d", maxAttempts)
}
rows, err := r.store.pool.Query(ctx, `
WITH candidates AS (
SELECT qte.id
FROM question_table_exports qte
WHERE (qte.status = 'pending'
OR (qte.status = 'export_failed' AND (qte.next_attempt_at IS NULL OR qte.next_attempt_at <= NOW()))
OR (qte.status = 'exporting' AND qte.last_attempt_at < $3))
AND qte.attempts < $2
ORDER BY qte.created_at, qte.id
LIMIT $1
FOR UPDATE SKIP LOCKED
),
claimed AS (
UPDATE question_table_exports qte
SET status = 'exporting',
attempts = qte.attempts + 1,
last_attempt_at = NOW(),
next_attempt_at = NULL,
failure_code = '',
updated_at = NOW()
FROM candidates c
WHERE qte.id = c.id
RETURNING qte.id, qte.attempts, qte.tenant_key, qte.channel, qte.turn_id, qte.created_at
)
SELECT
cl.id,
cl.attempts,
t.question,
t.answer,
m.created_at,
COALESCE(m.actor_name_snapshot, '') AS actor_name_snapshot,
COALESCE(m.actor_union_id_snapshot, '') AS actor_union_id_snapshot,
COALESCE(m.trigger_person_snapshot, '') AS trigger_person_snapshot,
COALESCE(m.department_snapshot, '') AS department_snapshot,
COALESCE(m.region_snapshot, '') AS region_snapshot
FROM claimed cl
JOIN turns t ON t.id = cl.turn_id
AND t.tenant_key = cl.tenant_key
AND t.channel = cl.channel
JOIN messages m ON m.id = t.request_message_id
AND m.tenant_key = cl.tenant_key
AND m.channel = cl.channel
ORDER BY cl.created_at, cl.id`,
limit, maxAttempts, staleBefore)
if err != nil {
return nil, fmt.Errorf("questiontable export claim: %w", err)
}
defer rows.Close()
var exports []questiontable.PendingExport
for rows.Next() {
var (
exportID string
attempts int
question string
answer string
askedTime time.Time
actorNameSnapshot string
actorUnionIDSnapshot string
triggerPersonSnap string
departmentSnap string
regionSnap string
)
if err := rows.Scan(&exportID, &attempts,
&question, &answer,
&askedTime,
&actorNameSnapshot, &actorUnionIDSnapshot,
&triggerPersonSnap, &departmentSnap, &regionSnap); err != nil {
rows.Close()
return nil, fmt.Errorf("questiontable export claim scan: %w", err)
}
exports = append(exports, questiontable.PendingExport{
ExportID: exportID,
Attempts: attempts,
Row: questiontable.Row{
Question: question,
Answer: answer,
AskedTime: askedTime,
AskerName: actorNameSnapshot,
AskerUnionID: actorUnionIDSnapshot,
TriggerPerson: triggerPersonSnap,
Department: departmentSnap,
Region: regionSnap,
},
})
}
if err := rows.Err(); err != nil {
return nil, fmt.Errorf("questiontable export claim rows: %w", err)
}
return exports, nil
}
// MarkExported implements questiontable.ExportRepository.
// It records the Notable external record ID for a successfully exported row.
// It transitions the row from exporting to exported. RowsAffected must be exactly 1.
func (r *QuestionTableExportRepository) MarkExported(ctx context.Context, exportID, externalRecordID string) error {
if strings.TrimSpace(exportID) == "" {
return errExportIDRequired
}
if strings.TrimSpace(externalRecordID) == "" {
return errLocationRequired
}
if r == nil || r.store == nil || r.store.pool == nil {
return errExportNilStore
}
tag, err := r.store.pool.Exec(ctx, `
UPDATE question_table_exports
SET status = 'exported',
external_record_id = $2,
exported_at = NOW(),
failure_code = '',
next_attempt_at = NULL,
updated_at = NOW()
WHERE id = $1 AND status = 'exporting'`,
exportID, externalRecordID)
if err != nil {
return fmt.Errorf("questiontable export mark exported: %w", err)
}
if tag.RowsAffected() != 1 {
return errExportRowsAffected
}
return nil
}
// MarkFailed implements questiontable.ExportRepository.
// It records a failure for a row in exporting status with a stable failure code.
// Only FailureCodeRemoteInsertFailed is accepted. The raw error is never persisted.
// A bounded retry delay is computed based on the current attempt count.
// RowsAffected must be exactly 1.
func (r *QuestionTableExportRepository) MarkFailed(ctx context.Context, exportID, failureCode string) error {
if strings.TrimSpace(exportID) == "" {
return errExportIDRequired
}
if failureCode != questiontable.FailureCodeRemoteInsertFailed {
return errBadExportFailure
}
if r == nil || r.store == nil || r.store.pool == nil {
return errExportNilStore
}
tag, err := r.store.pool.Exec(ctx, `
UPDATE question_table_exports
SET status = 'export_failed',
failure_code = $2,
next_attempt_at = NOW() + (
CASE
WHEN attempts <= 1 THEN '5 seconds'::interval
WHEN attempts = 2 THEN '15 seconds'::interval
WHEN attempts = 3 THEN '45 seconds'::interval
ELSE '5 minutes'::interval
END
),
updated_at = NOW()
WHERE id = $1 AND status = 'exporting'`,
exportID, failureCode)
if err != nil {
return fmt.Errorf("questiontable export mark failed: %w", err)
}
if tag.RowsAffected() != 1 {
return errExportRowsAffected
}
return nil
}
@@ -0,0 +1,437 @@
// Source-level contract tests for the question table export repository SQL.
// No real database connection is used.
package postgres
import (
"os"
"regexp"
"strings"
"testing"
)
func readQuestionTableExportsSource(t *testing.T) string {
t.Helper()
data, err := os.ReadFile("question_table_exports.go")
if err != nil {
t.Fatalf("read question_table_exports.go: %v", err)
}
return string(data)
}
func extractSQLFromQuestionTableExports(t *testing.T) []string {
t.Helper()
src := readQuestionTableExportsSource(t)
var sqls []string
re := regexp.MustCompile("`([^`]*)`")
for _, m := range re.FindAllStringSubmatch(src, -1) {
upper := strings.ToUpper(m[1])
if strings.Contains(upper, "SELECT") || strings.Contains(upper, "UPDATE") ||
strings.Contains(upper, "INSERT") || strings.Contains(upper, "DELETE") ||
strings.Contains(upper, "WITH") {
sqls = append(sqls, m[1])
}
}
return sqls
}
// ---------------------------------------------------------------------------
// Claim SQL structure
// ---------------------------------------------------------------------------
func TestQuestionTableExportsClaimUsesCTE(t *testing.T) {
sqls := extractSQLFromQuestionTableExports(t)
found := false
for _, sql := range sqls {
upper := strings.ToUpper(sql)
if strings.Contains(upper, "WITH CANDIDATES") {
found = true
break
}
}
if !found {
t.Fatal("Claim SQL with CTE not found in question_table_exports.go")
}
}
func TestQuestionTableExportsClaimUsesForUpdateSkipLocked(t *testing.T) {
sqls := extractSQLFromQuestionTableExports(t)
found := false
for _, sql := range sqls {
upper := strings.ToUpper(sql)
if strings.Contains(upper, "FOR UPDATE SKIP LOCKED") {
found = true
break
}
}
if !found {
t.Fatal("Claim must use FOR UPDATE SKIP LOCKED for atomicity")
}
}
func TestQuestionTableExportsClaimUsesSingleUpdate(t *testing.T) {
sqls := extractSQLFromQuestionTableExports(t)
var claimSQL string
for _, sql := range sqls {
upper := strings.ToUpper(sql)
if strings.Contains(upper, "WITH CANDIDATES") {
claimSQL = upper
break
}
}
if claimSQL == "" {
t.Fatal("Claim SQL with CTE not found")
}
updateCount := strings.Count(claimSQL, "UPDATE QUESTION_TABLE_EXPORTS")
if updateCount != 1 {
t.Fatalf("Claim must use exactly one UPDATE, found %d", updateCount)
}
}
// ---------------------------------------------------------------------------
// Claim eligibility: statuses
// ---------------------------------------------------------------------------
func TestQuestionTableExportsClaimStatusAndAttempts(t *testing.T) {
sqls := extractSQLFromQuestionTableExports(t)
joined := strings.ToUpper(strings.Join(sqls, " "))
// Status filter: pending, export_failed, or stale exporting
if !strings.Contains(joined, "STATUS = 'PENDING'") {
t.Error("Claim must include pending cases")
}
if !strings.Contains(joined, "STATUS = 'EXPORT_FAILED'") {
t.Error("Claim must include retryable export_failed cases")
}
if !strings.Contains(joined, "STATUS = 'EXPORTING'") {
t.Error("Claim must include stale exporting cases")
}
// Attempts check
if !strings.Contains(joined, "ATTEMPTS <") {
t.Error("Claim must filter by attempts < maxAttempts")
}
// Stale check
if !strings.Contains(joined, "LAST_ATTEMPT_AT <") {
t.Error("Claim must check last_attempt_at for stale reclaim")
}
}
// ---------------------------------------------------------------------------
// Claim: next_attempt_at condition for export_failed
// ---------------------------------------------------------------------------
func TestQuestionTableExportsClaimNextAttemptAt(t *testing.T) {
sqls := extractSQLFromQuestionTableExports(t)
joined := strings.ToUpper(strings.Join(sqls, " "))
if !strings.Contains(joined, "NEXT_ATTEMPT_AT IS NULL OR NEXT_ATTEMPT_AT <= NOW()") ||
!strings.Contains(joined, "NEXT_ATTEMPT_AT IS NULL") {
// At least check that next_attempt_at appears in the export_failed condition
if !strings.Contains(joined, "NEXT_ATTEMPT_AT") {
t.Error("Claim must use next_attempt_at for retry backoff on export_failed")
}
}
}
// ---------------------------------------------------------------------------
// Claim: order and limit
// ---------------------------------------------------------------------------
func TestQuestionTableExportsClaimOrderAndLimit(t *testing.T) {
sqls := extractSQLFromQuestionTableExports(t)
joined := strings.ToUpper(strings.Join(sqls, " "))
if !strings.Contains(joined, "ORDER BY") {
t.Error("Claim must ORDER BY created_at, id")
}
if !strings.Contains(joined, "LIMIT") {
t.Error("Claim must use parameterized LIMIT")
}
}
// ---------------------------------------------------------------------------
// Claim: scope joins with tenant_key and channel
// ---------------------------------------------------------------------------
func TestQuestionTableExportsClaimScopeJoins(t *testing.T) {
sqls := extractSQLFromQuestionTableExports(t)
joined := strings.ToUpper(strings.Join(sqls, " "))
// Must join on tenant_key and channel for scoping on turns
if !strings.Contains(joined, "T.TENANT_KEY = CL.TENANT_KEY") {
t.Error("Claim turns join must include tenant_key scope")
}
if !strings.Contains(joined, "T.CHANNEL = CL.CHANNEL") {
t.Error("Claim turns join must include channel scope")
}
// Must join on tenant_key and channel for scoping on messages
if !strings.Contains(joined, "M.TENANT_KEY = CL.TENANT_KEY") {
t.Error("Claim messages join must include tenant_key scope")
}
if !strings.Contains(joined, "M.CHANNEL = CL.CHANNEL") {
t.Error("Claim messages join must include channel scope")
}
}
// ---------------------------------------------------------------------------
// Claim: reads question/answer from turns, not stored in export table
// ---------------------------------------------------------------------------
func TestQuestionTableExportsClaimReadsQuestionAnswerFromTurns(t *testing.T) {
sqls := extractSQLFromQuestionTableExports(t)
joined := strings.ToUpper(strings.Join(sqls, " "))
// Must read question/answer from turns table
if !strings.Contains(joined, "T.QUESTION") {
t.Error("Claim must read question from turns table")
}
if !strings.Contains(joined, "T.ANSWER") {
t.Error("Claim must read answer from turns table")
}
}
// ---------------------------------------------------------------------------
// Claim: reads actor_union_id_snapshot from messages
// ---------------------------------------------------------------------------
func TestQuestionTableExportsClaimReadsActorUnionIDSnapshot(t *testing.T) {
sqls := extractSQLFromQuestionTableExports(t)
joined := strings.ToUpper(strings.Join(sqls, " "))
if !strings.Contains(joined, "ACTOR_UNION_ID_SNAPSHOT") {
t.Error("Claim must read actor_union_id_snapshot from messages")
}
}
// ---------------------------------------------------------------------------
// Claim: transition clears next_attempt_at
// ---------------------------------------------------------------------------
func TestQuestionTableExportsClaimClearsNextAttemptAt(t *testing.T) {
sqls := extractSQLFromQuestionTableExports(t)
joined := strings.ToUpper(strings.Join(sqls, " "))
if !strings.Contains(joined, "NEXT_ATTEMPT_AT = NULL") {
t.Error("Claim UPDATE must clear next_attempt_at")
}
}
// ---------------------------------------------------------------------------
// MarkExported guards
// ---------------------------------------------------------------------------
func TestQuestionTableExportsMarkExportedGuard(t *testing.T) {
sqls := extractSQLFromQuestionTableExports(t)
var markSQL string
for _, sql := range sqls {
upper := strings.ToUpper(sql)
if strings.Contains(upper, "SET STATUS = 'EXPORTED'") {
markSQL = sql
break
}
}
if markSQL == "" {
t.Fatal("MarkExported SQL not found")
}
upper := strings.ToUpper(markSQL)
// Must guard on status = 'exporting'
if !strings.Contains(upper, "STATUS = 'EXPORTING'") {
t.Error("MarkExported must guard on status = 'exporting'")
}
// Must set external_record_id
if !strings.Contains(upper, "EXTERNAL_RECORD_ID") {
t.Error("MarkExported must set external_record_id")
}
// Must clear failure_code
if !strings.Contains(upper, "FAILURE_CODE = ''") {
t.Error("MarkExported must clear failure_code")
}
// Must clear next_attempt_at
if !strings.Contains(upper, "NEXT_ATTEMPT_AT = NULL") {
t.Error("MarkExported must clear next_attempt_at")
}
}
func TestQuestionTableExportsMarkExportedRowsAffected(t *testing.T) {
src := readQuestionTableExportsSource(t)
if !strings.Contains(src, "RowsAffected()") {
t.Error("MarkExported must check RowsAffected()")
}
}
// ---------------------------------------------------------------------------
// MarkFailed guards
// ---------------------------------------------------------------------------
func TestQuestionTableExportsMarkFailedGuard(t *testing.T) {
sqls := extractSQLFromQuestionTableExports(t)
var markSQL string
for _, sql := range sqls {
upper := strings.ToUpper(sql)
if strings.Contains(upper, "SET STATUS = 'EXPORT_FAILED'") {
markSQL = sql
break
}
}
if markSQL == "" {
t.Fatal("MarkFailed SQL not found")
}
upper := strings.ToUpper(markSQL)
// Must guard on status = 'exporting'
if !strings.Contains(upper, "STATUS = 'EXPORTING'") {
t.Error("MarkFailed must guard on status = 'exporting'")
}
// Must set failure_code
if !strings.Contains(upper, "FAILURE_CODE") {
t.Error("MarkFailed must set failure_code")
}
}
func TestQuestionTableExportsMarkFailedOnlyAcceptsStableCode(t *testing.T) {
src := readQuestionTableExportsSource(t)
if !strings.Contains(src, "FailureCodeRemoteInsertFailed") {
t.Error("MarkFailed must validate failureCode == FailureCodeRemoteInsertFailed")
}
}
func TestQuestionTableExportsMarkFailedRowsAffected(t *testing.T) {
src := readQuestionTableExportsSource(t)
if !strings.Contains(src, "RowsAffected()") {
t.Error("MarkFailed must check RowsAffected()")
}
}
// ---------------------------------------------------------------------------
// MarkFailed: backoff intervals
// ---------------------------------------------------------------------------
func TestQuestionTableExportsMarkFailedBackoffIntervals(t *testing.T) {
sqls := extractSQLFromQuestionTableExports(t)
joined := strings.ToUpper(strings.Join(sqls, " "))
if !strings.Contains(joined, "5 SECONDS") {
t.Error("MarkFailed must use 5s backoff for attempts <= 1")
}
if !strings.Contains(joined, "15 SECONDS") {
t.Error("MarkFailed must use 15s backoff for attempts = 2")
}
if !strings.Contains(joined, "45 SECONDS") {
t.Error("MarkFailed must use 45s backoff for attempts = 3")
}
if !strings.Contains(joined, "5 MINUTES") {
t.Error("MarkFailed must use 5m backoff for attempts >= 4")
}
}
// ---------------------------------------------------------------------------
// No raw error leak in SQL
// ---------------------------------------------------------------------------
func TestQuestionTableExportsNoRawErrorLeak(t *testing.T) {
sqls := extractSQLFromQuestionTableExports(t)
joined := strings.ToUpper(strings.Join(sqls, " "))
for _, forbidden := range []string{
"TOKEN", "WEBHOOK", "RAW_ERROR", "RAW_RESPONSE",
"GROUP_ID", "EXTERNAL_CHAT_ID", "EXTERNAL_USER_ID",
} {
if strings.Contains(joined, forbidden) {
t.Errorf("question table export SQL must not contain %q", forbidden)
}
}
}
// ---------------------------------------------------------------------------
// No question/answer bodies stored in export table
// ---------------------------------------------------------------------------
func TestQuestionTableExportsNoQuestionAnswerBodiesStored(t *testing.T) {
sqls := extractSQLFromQuestionTableExports(t)
// Only check standalone MarkExported/MarkFailed UPDATE statements, not the CTE
// which has a SELECT with t.question/t.answer (that's fine — they come from turns).
for _, sql := range sqls {
upper := strings.ToUpper(sql)
// Skip CTE SQL (contains WITH CANDIDATES)
if strings.Contains(upper, "WITH CANDIDATES") {
continue
}
if strings.Contains(upper, "UPDATE QUESTION_TABLE_EXPORTS") {
if strings.Contains(upper, "QUESTION") && !strings.Contains(upper, "QUESTION_TABLE_EXPORTS") {
t.Errorf("UPDATE must not set question column: %q", sql)
}
if strings.Contains(upper, "ANSWER") {
t.Errorf("UPDATE must not set answer column: %q", sql)
}
}
}
}
// ---------------------------------------------------------------------------
// Nil safety
// ---------------------------------------------------------------------------
func TestQuestionTableExportsNilSafe(t *testing.T) {
src := readQuestionTableExportsSource(t)
count := strings.Count(src, "r.store == nil || r.store.pool == nil")
if count < 3 {
t.Errorf("question table export methods must guard against nil store/pool, found %d guards", count)
}
}
// ---------------------------------------------------------------------------
// ExportID validation
// ---------------------------------------------------------------------------
func TestQuestionTableExportsExportIDValidation(t *testing.T) {
src := readQuestionTableExportsSource(t)
if !strings.Contains(src, "TrimSpace(exportID)") {
t.Error("MarkExported/MarkFailed must validate exportID is not empty")
}
}
// ---------------------------------------------------------------------------
// Adapter: compile-time assertion and constructor
// ---------------------------------------------------------------------------
func TestQuestionTableExportsCompileTimeAssertion(t *testing.T) {
src := readQuestionTableExportsSource(t)
if !strings.Contains(src, "var _ questiontable.ExportRepository = (*QuestionTableExportRepository)(nil)") {
t.Error("QuestionTableExportRepository must have compile-time interface assertion")
}
}
func TestQuestionTableExportsConstructorNilSafety(t *testing.T) {
src := readQuestionTableExportsSource(t)
if !strings.Contains(src, "NewQuestionTableExportRepository") {
t.Error("Must have a validated constructor NewQuestionTableExportRepository")
}
if !strings.Contains(src, "errExportNilStore") {
t.Error("Constructor must reject nil store with a stable error")
}
}
// ---------------------------------------------------------------------------
// Claim: input validation
// ---------------------------------------------------------------------------
func TestQuestionTableExportsClaimInputValidation(t *testing.T) {
src := readQuestionTableExportsSource(t)
if !strings.Contains(src, "limit <= 0") || !strings.Contains(src, "limit > 1000") {
t.Error("Claim must validate positive bounded limit")
}
if !strings.Contains(src, "maxAttempts <= 0") || !strings.Contains(src, "maxAttempts > 100") {
t.Error("Claim must validate positive bounded maxAttempts")
}
}
// ---------------------------------------------------------------------------
// MarkExported: external_record_id validation
// ---------------------------------------------------------------------------
func TestQuestionTableExportsMarkExportedLocationValidation(t *testing.T) {
src := readQuestionTableExportsSource(t)
if !strings.Contains(src, "TrimSpace(externalRecordID)") {
t.Error("MarkExported must validate externalRecordID is not empty")
}
}
// ---------------------------------------------------------------------------
// No a1_notation references
// ---------------------------------------------------------------------------
func TestQuestionTableExportsNoA1Notation(t *testing.T) {
src := readQuestionTableExportsSource(t)
if strings.Contains(src, "a1_notation") {
t.Error("question_table_exports.go must not reference stale a1_notation column")
}
}
+28 -6
View File
@@ -83,13 +83,13 @@ func (s *Store) BeginTurn(ctx context.Context, turn core.Turn) (core.Turn, bool,
if err = tx.QueryRow(ctx, `INSERT INTO messages
(id,tenant_key,channel,conversation_id,actor_id,external_message_id,correlation_id,
direction,role,message_type,text_content,media_json,status,actor_name_snapshot,department_snapshot,region_snapshot,safe_metadata_json)
VALUES($1,$2,$3,$4,$5,$6,$7,'inbound','user',$8,$9,$10,'received',$11,$12,$13,'{}'::JSONB)
direction,role,message_type,text_content,media_json,status,actor_name_snapshot,trigger_person_snapshot,actor_union_id_snapshot,department_snapshot,region_snapshot,safe_metadata_json)
VALUES($1,$2,$3,$4,$5,$6,$7,'inbound','user',$8,$9,$10,'received',$11,$12,$13,$14,$15,'{}'::JSONB)
ON CONFLICT (tenant_key,channel,external_message_id) WHERE external_message_id!=''
DO UPDATE SET updated_at=messages.updated_at
RETURNING id`, turn.RequestMessageID, turn.TenantKey, turn.Channel, turn.ConversationID,
turn.ActorID, turn.ExternalMessageID, turn.ID, turn.MessageType, turn.Question, mediaJSON,
turn.SenderName, turn.DepartmentSnapshot, turn.RegionSnapshot).Scan(&turn.RequestMessageID); err != nil {
turn.SenderName, turn.TriggerPersonSnapshot, turn.ActorUnionIDSnapshot, turn.DepartmentSnapshot, turn.RegionSnapshot).Scan(&turn.RequestMessageID); err != nil {
return core.Turn{}, false, err
}
@@ -207,6 +207,21 @@ func (s *Store) FinishTurn(ctx context.Context, turn core.Turn) error {
return err
}
answerMessageID = turn.AnswerMessageID
if shouldEnqueueQuestionTableExport(turn) {
exportID, err := newUUID()
if err != nil {
return err
}
_, err = tx.Exec(ctx, `INSERT INTO question_table_exports
(id,tenant_key,channel,turn_id,status)
VALUES($1,$2,$3,$4,'pending')
ON CONFLICT (tenant_key,channel,turn_id) DO NOTHING`,
exportID, turn.TenantKey, turn.Channel, turn.ID)
if err != nil {
return err
}
}
}
_, err = tx.Exec(ctx, `UPDATE turns SET weknora_session_id=$2,weknora_request_id=$3,answer=$4,references_json=$5,media_json=$6,status=$7,error_message=$8,answer_message_id=COALESCE($9::TEXT,answer_message_id),updated_at=NOW() WHERE id=$1`,
turn.ID, turn.WeKnoraSessionID, turn.WeKnoraRequestID, turn.Answer, references, mediaJSON,
@@ -264,14 +279,14 @@ func loadTurnWith(ctx context.Context, query rowQuerier, tenantKey, channel, mes
var references []byte
var mediaJSON []byte
var agentID, requestMessageID, answerMessageID *string
var actorID, conversationID, messageType, actorName *string
var actorID, conversationID, messageType, actorName, actorUnionID *string
var conversationType, externalConversationID, externalActorID *string
var answerDeliveryStatus, answerDeliveryMethod *string
err := query.QueryRow(ctx, `SELECT
t.id,t.tenant_key,t.channel,t.external_message_id,t.session_key,t.weknora_session_id,
t.weknora_request_id,t.question,t.answer,t.references_json,t.media_json,t.status,t.error_message,
t.agent_id,t.request_message_id,t.answer_message_id,
m.actor_id,m.conversation_id,m.message_type,m.actor_name_snapshot,
m.actor_id,m.conversation_id,m.message_type,m.actor_name_snapshot,m.actor_union_id_snapshot,
c.conversation_type,c.external_conversation_id,a.external_actor_id,
am.status,am.delivery_method
FROM turns t
@@ -283,7 +298,7 @@ func loadTurnWith(ctx context.Context, query rowQuerier, tenantKey, channel, mes
&turn.ID, &turn.TenantKey, &turn.Channel, &turn.ExternalMessageID, &turn.SessionKey,
&turn.WeKnoraSessionID, &turn.WeKnoraRequestID, &turn.Question, &turn.Answer, &references,
&mediaJSON, &turn.Status, &turn.ErrorMessage, &agentID, &requestMessageID, &answerMessageID,
&actorID, &conversationID, &messageType, &actorName, &conversationType,
&actorID, &conversationID, &messageType, &actorName, &actorUnionID, &conversationType,
&externalConversationID, &externalActorID, &answerDeliveryStatus, &answerDeliveryMethod)
if err != nil {
return core.Turn{}, err
@@ -315,6 +330,9 @@ func loadTurnWith(ctx context.Context, query rowQuerier, tenantKey, channel, mes
if actorName != nil {
turn.SenderName = *actorName
}
if actorUnionID != nil {
turn.ActorUnionIDSnapshot = *actorUnionID
}
if externalActorID != nil {
turn.ExternalUserID = *externalActorID
}
@@ -343,6 +361,10 @@ func shouldCreateAnswerMessage(turn core.Turn) bool {
strings.TrimSpace(turn.AnswerMessageID) != ""
}
func shouldEnqueueQuestionTableExport(turn core.Turn) bool {
return shouldCreateAnswerMessage(turn) && strings.EqualFold(strings.TrimSpace(turn.Channel), "dingtalk")
}
func validateDeliveryUpdate(update core.DeliveryUpdate) error {
if strings.TrimSpace(update.TenantKey) == "" || strings.TrimSpace(update.Channel) == "" ||
strings.TrimSpace(update.AnswerMessageID) == "" || strings.TrimSpace(update.CorrelationID) == "" {