ai-customer-service/test/e2e/sub2api_callback_flow_test.go

package e2e

import (
	"bytes"
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"net/http"
	"net/http/httptest"
	"testing"
	"time"

	"github.com/bridge/ai-customer-service/internal/app"
	"github.com/bridge/ai-customer-service/internal/config"
	"github.com/bridge/ai-customer-service/internal/domain/platformevent"
	"github.com/bridge/ai-customer-service/internal/http/handlers"
	"github.com/bridge/ai-customer-service/internal/platform/logging"
	pgstore "github.com/bridge/ai-customer-service/internal/store/postgres"
)

func e2ePlatformDSN() string {
	return "host=localhost port=5434 user=ai_cs password=ai_cs_secret dbname=ai_customer_service sslmode=disable"
}

func openE2EPlatformDB(t *testing.T) *sql.DB {
	t.Helper()
	db, err := pgstore.Open(pgstore.Config{
		DSN:             e2ePlatformDSN(),
		MaxOpenConns:    5,
		MaxIdleConns:    2,
		ConnMaxLifetime: 30 * time.Second,
	})
	if err != nil {
		t.Skipf("PostgreSQL not available, skipping E2E test: %v", err)
	}
	if err := pgstore.RunMigrations(db, "../../db/migration"); err != nil {
		_ = db.Close()
		t.Fatalf("run migrations failed: %v", err)
	}
	return db
}

func resetE2EPlatformDB(t *testing.T, db *sql.DB) {
	t.Helper()
	if db == nil {
		t.Fatal("db is nil")
	}
	if _, err := db.ExecContext(context.Background(), `
		TRUNCATE TABLE
			cs_platform_event_dead_letters,
			cs_platform_event_delivery_attempts,
			cs_platform_event_outbox,
			cs_message_dedup,
			cs_messages,
			cs_tickets,
			cs_audit_logs,
			cs_sessions
		RESTART IDENTITY CASCADE
	`); err != nil {
		t.Fatalf("reset e2e postgres state failed: %v", err)
	}
}

func newSub2APIE2EApp(t *testing.T, db *sql.DB, callbackURL string, callbackSecret string, maxRetries int) *app.App {
	t.Helper()
	cfg := &config.Config{}
	cfg.HTTP.Addr = ":0"
	cfg.HTTP.ReadHeaderTimeout = 5
	cfg.HTTP.ReadTimeout = 10
	cfg.HTTP.WriteTimeout = 15
	cfg.HTTP.IdleTimeout = 60
	cfg.HTTP.MaxHeaderBytes = 1 << 20
	cfg.HTTP.MaxBodyBytes = 1 << 20
	cfg.Runtime.Env = "test"
	cfg.Postgres.Enabled = true
	cfg.Postgres.DSN = e2ePlatformDSN()
	cfg.Postgres.MigrationDir = "../../db/migration"
	cfg.Postgres.MaxOpenConns = 5
	cfg.Postgres.MaxIdleConns = 2
	cfg.Postgres.ConnMaxLifetime = 30
	cfg.Webhook.Secret = "default-webhook-secret"
	cfg.Webhook.TimestampHeader = "X-CS-Timestamp"
	cfg.Webhook.SignatureHeader = "X-CS-Signature"
	cfg.Webhook.MaxSkewSeconds = 300
	cfg.PlatformAdapters.Enabled = true
	cfg.PlatformAdapters.Sub2API.Enabled = true
	cfg.PlatformAdapters.Sub2API.IngressSecret = "sub2api-ingress-secret"
	cfg.PlatformAdapters.Sub2API.CallbackBaseURL = callbackURL
	cfg.PlatformAdapters.Sub2API.CallbackSecret = callbackSecret
	cfg.PlatformAdapters.Sub2API.CallbackTimeoutMS = 2000
	cfg.PlatformAdapters.Sub2API.CallbackMaxRetries = maxRetries
	cfg.PlatformAdapters.Sub2API.CallbackPollIntervalMS = 200
	cfg.PlatformAdapters.Sub2API.CallbackBatchSize = 8
	cfg.PlatformAdapters.Sub2API.CallbackRetrySchedule = []int{1, 2, 5}

	application, err := app.New(cfg, logging.New())
	if err != nil {
		t.Fatalf("app.New() error = %v", err)
	}
	t.Cleanup(func() {
		_ = application.Shutdown(context.Background())
		if db != nil {
			_ = db.Close()
		}
	})
	return application
}

func eventually(t *testing.T, timeout time.Duration, fn func() bool) {
	t.Helper()
	deadline := time.Now().Add(timeout)
	for time.Now().Before(deadline) {
		if fn() {
			return
		}
		time.Sleep(200 * time.Millisecond)
	}
	t.Fatal("condition not satisfied before timeout")
}

func waitForSessionEvents(t *testing.T, timeout time.Duration, eventsCh <-chan platformevent.Event, sessionID string, want int) []platformevent.Event {
	t.Helper()
	deadline := time.Now().Add(timeout)
	var filtered []platformevent.Event
	for time.Now().Before(deadline) {
		select {
		case event := <-eventsCh:
			if event.SessionID == sessionID {
				filtered = append(filtered, event)
			}
			if len(filtered) == want {
				return filtered
			}
		case <-time.After(50 * time.Millisecond):
		}
	}

	snapshot := make([]string, 0)
	for {
		select {
		case event := <-eventsCh:
			snapshot = append(snapshot, fmt.Sprintf("%s/%s", event.SessionID, event.EventType))
		default:
		}
		break
	}
	t.Fatalf("session %s received %d events, want %d; snapshot=%v", sessionID, len(filtered), want, snapshot)
	return nil
}

func TestSub2APICallbackFlow_ShouldDeliverOrderedEventsWithStableEventIDs(t *testing.T) {
	db := openE2EPlatformDB(t)
	resetE2EPlatformDB(t, db)

	eventsCh := make(chan platformevent.Event)
	callbackServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		var event platformevent.Event
		if err := json.NewDecoder(r.Body).Decode(&event); err != nil {
			t.Fatalf("decode callback body failed: %v", err)
		}
		select {
		case eventsCh <- event:
		case <-time.After(5 * time.Second):
			t.Fatalf("eventsCh send timeout")
		}
		w.WriteHeader(http.StatusOK)
	}))
	defer callbackServer.Close()

	application := newSub2APIE2EApp(t, db, callbackServer.URL, "sub2api-callback-secret", 3)
	server := httptest.NewServer(application.Server.Handler)
	defer server.Close()

	openID := "sub2api-e2e-" + time.Now().UTC().Format("150405.000000000")
	payload := map[string]any{
		"message_id": "m-e2e-" + time.Now().UTC().Format("150405.000000000"),
		"channel":    "sub2api",
		"open_id":    openID,
		"content":    "我要退款",
	}
	body, _ := json.Marshal(payload)
	timestamp, signature, err := handlers.SignWebhookRequest("sub2api-ingress-secret", time.Now().Unix(), body)
	if err != nil {
		t.Fatalf("SignWebhookRequest() error = %v", err)
	}
	req, err := http.NewRequest(http.MethodPost, server.URL+"/api/v1/customer-service/platforms/sub2api/webhook", bytes.NewReader(body))
	if err != nil {
		t.Fatalf("new request error = %v", err)
	}
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("X-CS-Timestamp", timestamp)
	req.Header.Set("X-CS-Signature", signature)
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatalf("do request error = %v", err)
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("status = %d, want 200", resp.StatusCode)
	}
	var ack map[string]any
	if err := json.NewDecoder(resp.Body).Decode(&ack); err != nil {
		t.Fatalf("decode ack failed: %v", err)
	}
	sessionID, _ := ack["session_id"].(string)
	if sessionID == "" {
		t.Fatalf("ack session_id = %v, want non-empty", ack["session_id"])
	}

	filtered := waitForSessionEvents(t, 8*time.Second, eventsCh, sessionID, 6)

	wantTypes := []string{
		platformevent.TypeMessageReceived,
		platformevent.TypeMessageProcessing,
		platformevent.TypeIntentResolved,
		platformevent.TypeHandoffTriggered,
		platformevent.TypeTicketCreated,
		platformevent.TypeReplyGenerated,
	}
	seenIDs := make(map[string]struct{}, len(filtered))
	for i, event := range filtered {
		if event.EventType != wantTypes[i] {
			t.Fatalf("event[%d].type = %s, want %s", i, event.EventType, wantTypes[i])
		}
		if event.ID == "" {
			t.Fatalf("event[%d] id is empty", i)
		}
		if _, exists := seenIDs[event.ID]; exists {
			t.Fatalf("duplicate event id: %s", event.ID)
		}
		seenIDs[event.ID] = struct{}{}
	}

	var deliveredCount int
	eventually(t, 8*time.Second, func() bool {
		err := db.QueryRowContext(context.Background(), `
			SELECT COUNT(1)
			FROM cs_platform_event_outbox
			WHERE platform = 'sub2api' AND status = 'delivered' AND session_id IN (
				SELECT id FROM cs_sessions WHERE channel = 'sub2api' AND open_id = $1
			)
		`, openID).Scan(&deliveredCount)
		if err != nil {
			t.Fatalf("query delivered count failed: %v", err)
		}
		return deliveredCount == 6
	})
}

func TestSub2APICallbackFlow_ShouldDeadLetterAfterMaxRetries(t *testing.T) {
	db := openE2EPlatformDB(t)
	resetE2EPlatformDB(t, db)

	callbackServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
		w.WriteHeader(http.StatusBadGateway)
		_, _ = w.Write([]byte(`{"error":"upstream"}`))
	}))
	defer callbackServer.Close()

	application := newSub2APIE2EApp(t, db, callbackServer.URL, "sub2api-callback-secret", 1)
	server := httptest.NewServer(application.Server.Handler)
	defer server.Close()

	openID := "sub2api-dead-" + time.Now().UTC().Format("150405.000000000")
	payload := map[string]any{
		"message_id": "m-dead-" + time.Now().UTC().Format("150405.000000000"),
		"channel":    "sub2api",
		"open_id":    openID,
		"content":    "晚上好",
	}
	body, _ := json.Marshal(payload)
	timestamp, signature, err := handlers.SignWebhookRequest("sub2api-ingress-secret", time.Now().Unix(), body)
	if err != nil {
		t.Fatalf("SignWebhookRequest() error = %v", err)
	}
	req, err := http.NewRequest(http.MethodPost, server.URL+"/api/v1/customer-service/platforms/sub2api/webhook", bytes.NewReader(body))
	if err != nil {
		t.Fatalf("new request error = %v", err)
	}
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("X-CS-Timestamp", timestamp)
	req.Header.Set("X-CS-Signature", signature)
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatalf("do request error = %v", err)
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("status = %d, want 200", resp.StatusCode)
	}

	eventually(t, 8*time.Second, func() bool {
		var deadCount int
		err := db.QueryRowContext(context.Background(), `
			SELECT COUNT(1)
			FROM cs_platform_event_dead_letters dl
			JOIN cs_platform_event_outbox o ON o.id = dl.event_id
			WHERE o.platform = 'sub2api' AND o.session_id IN (
				SELECT id FROM cs_sessions WHERE channel = 'sub2api' AND open_id = $1
			)
		`, openID).Scan(&deadCount)
		return err == nil && deadCount == 4
	})
}