package lifecycle import ( "context" "fmt" "os" "sync" "testing" "time" "git.ilapage.cn/ila/yovision/Bell/server/app/alert/query" "git.ilapage.cn/ila/yovision/Bell/server/app/audit" "git.ilapage.cn/ila/yovision/Bell/server/app/auth" "git.ilapage.cn/ila/yovision/Bell/server/app/event" "git.ilapage.cn/ila/yovision/Bell/server/app/rule" "git.ilapage.cn/ila/yovision/Bell/server/internal/platform" "git.ilapage.cn/ila/yovision/Bell/server/migrations" ) func TestConcurrentAckCloseAndRestart(t *testing.T) { url := os.Getenv("BELL_TEST_DATABASE_URL") if url == "" { t.Skip("BELL_TEST_DATABASE_URL is not set") } ctx := context.Background() db, err := platform.OpenDatabase(ctx, url) if err != nil { t.Fatal(err) } if err := migrations.Apply(ctx, db); err != nil { t.Fatal(err) } prefix := fmt.Sprintf("lifecycle-%d", time.Now().UnixNano()) actors := make([]auth.User, 2) for i := range actors { var id string username := fmt.Sprintf("%s-%d", prefix, i) if err := db.QueryRow(ctx, `INSERT INTO bell_users(username,display_name,password_hash) VALUES($1,$2,'test-only-not-a-login-hash') RETURNING id::text`, username, username).Scan(&id); err != nil { t.Fatal(err) } if _, err := db.Exec(ctx, `INSERT INTO bell_user_roles(user_id,role_id) SELECT $1,id FROM bell_roles WHERE code='operator'`, id); err != nil { t.Fatal(err) } actors[i] = auth.User{ID: id, Username: username, DisplayName: username, Roles: []string{"operator"}} } eventType := prefix + "-event" ruleService := rule.Service{DB: db} createdRule, err := ruleService.Create(ctx, rule.CreateInput{Code: prefix, Name: "生命周期测试规则", EventType: &eventType, MinimumSeverity: "low"}) if err != nil { t.Fatal(err) } eventService := event.Service{DB: db, AfterPersist: ruleService.Match} accepted, err := eventService.Ingest(ctx, event.Command{ProducerID: prefix, SourceEventID: "1", EventType: eventType, OccurredAt: time.Now().UTC(), Location: "test place", Severity: "high", Attributes: map[string]any{}}) if err != nil { t.Fatal(err) } queryService := query.Service{DB: db} alerts, err := queryService.ForEvent(ctx, accepted.Event.ID) if err != nil { t.Fatal(err) } alertID := "" for _, item := range alerts { if item.PrimaryRuleID == createdRule.ID { alertID = item.ID } } if alertID == "" { t.Fatal("alert not created") } service := Service{DB: db, Audit: audit.Store{DB: db}, Query: queryService} const attempts = 20 var wg sync.WaitGroup for i := 0; i < attempts; i++ { wg.Add(1) actor := actors[i%2] go func() { defer wg.Done(); _, _ = service.Ack(ctx, alertID, actor) }() } wg.Wait() detail, err := queryService.Get(ctx, alertID) if err != nil { t.Fatal(err) } if detail.Alert.Status != "acknowledged" || detail.Alert.AcknowledgedBy == nil { t.Fatalf("unexpected ack projection: %#v", detail.Alert) } var facts int if err := db.QueryRow(ctx, `SELECT count(*) FROM bell_alert_lifecycle_facts WHERE alert_id=$1 AND transition='acknowledged'`, alertID).Scan(&facts); err != nil { t.Fatal(err) } if facts != 1 { t.Fatalf("ack facts=%d", facts) } winner := actors[0] loser := actors[1] if winner.ID != *detail.Alert.AcknowledgedBy { winner, loser = loser, winner } if _, err := service.Close(ctx, alertID, CloseInput{Outcome: "site_normal"}, loser); err == nil { t.Fatal("non-owner closed alert") } closed, err := service.Close(ctx, alertID, CloseInput{Outcome: "site_normal", Note: "现场检查正常"}, winner) if err != nil { t.Fatal(err) } if closed.Detail.Alert.Status != "closed" { t.Fatal("alert not closed") } repeat, err := service.Close(ctx, alertID, CloseInput{Outcome: "site_normal", Note: "现场检查正常"}, winner) if err != nil || !repeat.Idempotent { t.Fatalf("repeat close: %#v %v", repeat, err) } if err := db.QueryRow(ctx, `SELECT count(*) FROM bell_alert_lifecycle_facts WHERE alert_id=$1`, alertID).Scan(&facts); err != nil { t.Fatal(err) } if facts != 2 { t.Fatalf("lifecycle facts=%d", facts) } if _, err := db.Exec(ctx, `DELETE FROM bell_alert_lifecycle_facts WHERE alert_id=$1`, alertID); err == nil { t.Fatal("lifecycle deletion succeeded") } db.Close() reopened, err := platform.OpenDatabase(ctx, url) if err != nil { t.Fatal(err) } defer reopened.Close() afterRestart, err := (query.Service{DB: reopened}).Get(ctx, alertID) if err != nil { t.Fatal(err) } if afterRestart.Alert.Status != "closed" || len(afterRestart.Timeline) != 2 { t.Fatalf("restart projection: %#v", afterRestart) } }