Agent-System/internal/notification/service_test.go

package notification_test

import (
	"context"
	"os"
	"strings"
	"testing"
	"time"

	"github.com/jackc/pgx/v5/pgxpool"

	"chat-agent-system/internal/db"
	"chat-agent-system/internal/notification"
)

func openPool(t *testing.T) *pgxpool.Pool {
	t.Helper()
	// 隔离约定：仅提供专用测试库时运行，禁止默认回退到开发库。
	if os.Getenv("TEST_DATABASE_URL") == "" {
		t.Skip("TEST_DATABASE_URL not set; skipping notification PG test")
		return nil
	}
	dsn := os.Getenv("TEST_DATABASE_URL")
	if strings.Contains(dsn, "agent_system_dev") || strings.Contains(dsn, "agent_system_test") || strings.Contains(dsn, "agent_system_main") {
		t.Fatal("TEST_DATABASE_URL must not point at a shared env database")
	}
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	pool, err := pgxpool.New(ctx, dsn)
	if err != nil {
		t.Skip(err)
	}
	if err := pool.Ping(ctx); err != nil {
		pool.Close()
		t.Skip(err)
	}
	migCtx, migCancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer migCancel()
	if err := db.RunMigrations(migCtx, pool, db.ResolveMigrationsDir()); err != nil {
		pool.Close()
		t.Skipf("migration failed: %v", err)
	}
	return pool
}

func TestPG_EmitStageReadyDedupe(t *testing.T) {
	pool := openPool(t)
	defer pool.Close()

	svc := notification.NewService(pool)
	if !svc.Available() {
		t.Fatal("notification service should be available after migration")
	}

	ctx := context.Background()
	var projectID, instanceID, stageInstID int64
	code := "notify-" + time.Now().Format("150405.000000")
	err := pool.QueryRow(ctx, `
		INSERT INTO projects (code, name, status, creator_user_id, lifecycle_status)
		VALUES ($1, 'Notify Test ' || $2, 'active', 2, 'active') RETURNING id
	`, code, code).Scan(&projectID)
	if err != nil {
		t.Fatalf("insert project: %v", err)
	}
	t.Cleanup(func() {
		_, _ = pool.Exec(context.Background(), `DELETE FROM projects WHERE id = $1`, projectID)
	})

	var wfVersionID int64
	err = pool.QueryRow(ctx, `SELECT id FROM workflow_versions ORDER BY id DESC LIMIT 1`).Scan(&wfVersionID)
	if err != nil {
		t.Skip("no workflow version for notification test")
	}

	err = pool.QueryRow(ctx, `
		INSERT INTO project_workflow_instances (project_id, workflow_version_id, instance_no, status, config_revision, plan_hash, config_snapshot, created_by)
		VALUES ($1, $2, 1, 'active', 3, 'h', '{}'::jsonb, 2) RETURNING id
	`, projectID, wfVersionID).Scan(&instanceID)
	if err != nil {
		t.Fatalf("insert instance: %v", err)
	}

	err = pool.QueryRow(ctx, `
		INSERT INTO project_stage_instances (workflow_instance_id, stage_key, position, enabled, status)
		VALUES ($1, 'biz_intake', 0, true, 'active') RETURNING id
	`, instanceID).Scan(&stageInstID)
	if err != nil {
		t.Fatalf("insert stage: %v", err)
	}

	payload := notification.StageReadyPayload{
		ProjectID:          projectID,
		WorkflowInstanceID: instanceID,
		StageInstanceID:    stageInstID,
		StageKey:           "biz_intake",
		StageName:          "商务摸底",
		ConfigRevision:     3,
		RequiredDocCount:   1,
		ApprovedDocCount:   1,
	}

	created, err := svc.EmitStageReadyToAdvance(ctx, 2, payload)
	if err != nil {
		t.Fatalf("emit: %v", err)
	}
	if !created {
		t.Fatal("expected first emit to create notification")
	}

	again, err := svc.EmitStageReadyToAdvance(ctx, 2, payload)
	if err != nil {
		t.Fatalf("second emit: %v", err)
	}
	if again {
		t.Fatal("expected dedupe to suppress second notification")
	}

	var count int
	err = pool.QueryRow(ctx, `
		SELECT COUNT(*) FROM notifications WHERE user_id = 2 AND project_id = $1 AND type = $2
	`, projectID, notification.TypeStageReadyToAdvance).Scan(&count)
	if err != nil {
		t.Fatalf("count notifications: %v", err)
	}
	if count != 1 {
		t.Fatalf("expected exactly one notification, got %d", count)
	}
}