feat: enforce Bell alert lifecycle concurrency (#20)

This commit is contained in:
QiuSW
2026-08-12 18:09:52 +08:00
parent d55026902f
commit 10225f4613
10 changed files with 420 additions and 20 deletions
+58
View File
@@ -0,0 +1,58 @@
package lifecycle
import (
"encoding/json"
"errors"
"net/http"
"git.ilapage.cn/ila/yovision/Bell/server/app/auth"
"git.ilapage.cn/ila/yovision/Bell/server/app/rbac"
)
type HTTP struct {
Service Service
Auth auth.HTTP
}
func (h HTTP) Register(mux *http.ServeMux) {
mux.Handle("POST /api/v1/alerts/{id}/ack", h.Auth.Require(rbac.AlertsHandle, http.HandlerFunc(h.ack)))
mux.Handle("POST /api/v1/alerts/{id}/close", h.Auth.Require(rbac.AlertsHandle, http.HandlerFunc(h.close)))
}
func (h HTTP) ack(w http.ResponseWriter, r *http.Request) {
result, err := h.Service.Ack(r.Context(), r.PathValue("id"), auth.Principal(r.Context()))
if err != nil {
status := 500
if errors.Is(err, ErrAlreadyHandled) {
status = 409
}
writeJSON(w, status, map[string]any{"error": err.Error(), "current": result.Detail})
return
}
writeJSON(w, 200, result)
}
func (h HTTP) close(w http.ResponseWriter, r *http.Request) {
var input CloseInput
decoder := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<20))
decoder.DisallowUnknownFields()
if err := decoder.Decode(&input); err != nil {
writeJSON(w, 400, map[string]string{"error": "请求格式无效"})
return
}
result, err := h.Service.Close(r.Context(), r.PathValue("id"), input, auth.Principal(r.Context()))
if err != nil {
status := 500
if errors.Is(err, ErrOutcomeRequired) {
status = 400
} else if errors.Is(err, ErrInvalidTransition) {
status = 409
}
writeJSON(w, status, map[string]any{"error": err.Error(), "current": result.Detail})
return
}
writeJSON(w, 200, result)
}
func writeJSON(w http.ResponseWriter, status int, value any) {
w.Header().Set("Content-Type", "application/json; charset=utf-8")
w.WriteHeader(status)
_ = json.NewEncoder(w).Encode(value)
}
+128
View File
@@ -0,0 +1,128 @@
package lifecycle
import (
"context"
"encoding/json"
"errors"
"git.ilapage.cn/ila/yovision/Bell/server/app/alert/model"
alertQuery "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"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgxpool"
)
var ErrAlreadyHandled = errors.New("预警已由其他人员开始处理")
var ErrInvalidTransition = errors.New("当前状态不能执行此操作")
var ErrOutcomeRequired = errors.New("请选择现场处理结果")
type CloseInput struct {
Outcome string `json:"outcome"`
Note string `json:"note"`
}
type Result struct {
Detail model.Detail `json:"detail"`
Idempotent bool `json:"idempotent"`
}
type Service struct {
DB *pgxpool.Pool
Audit audit.Store
Query alertQuery.Service
}
func hasRole(user auth.User, role string) bool {
for _, value := range user.Roles {
if value == role {
return true
}
}
return false
}
func validOutcome(value string) bool {
return value == "danger_confirmed" || value == "false_positive" || value == "site_normal" || value == "unable_to_confirm"
}
func (s Service) Ack(ctx context.Context, alertID string, actor auth.User) (Result, error) {
tx, err := s.DB.Begin(ctx)
if err != nil {
return Result{}, err
}
var updated string
err = tx.QueryRow(ctx, `UPDATE bell_alerts SET status='acknowledged',acknowledged_by=$2,acknowledged_at=now(),updated_at=now() WHERE id=$1 AND status='open' RETURNING id::text`, alertID, actor.ID).Scan(&updated)
if err == nil {
if _, err = tx.Exec(ctx, `INSERT INTO bell_alert_lifecycle_facts(alert_id,transition,actor_user_id) VALUES($1,'acknowledged',$2)`, alertID, actor.ID); err == nil {
_, err = tx.Exec(ctx, `INSERT INTO bell_audit_log(actor_user_id,action,target_type,target_id,outcome,details) VALUES($1,'alert.ack','alert',$2,'success','{}')`, actor.ID, alertID)
}
if err == nil {
err = tx.Commit(ctx)
}
if err != nil {
_ = tx.Rollback(ctx)
return Result{}, err
}
detail, err := s.Query.Get(ctx, alertID)
return Result{Detail: detail}, err
}
_ = tx.Rollback(ctx)
if !errors.Is(err, pgx.ErrNoRows) {
return Result{}, err
}
detail, getErr := s.Query.Get(ctx, alertID)
if getErr != nil {
return Result{}, getErr
}
if detail.Alert.AcknowledgedBy != nil && *detail.Alert.AcknowledgedBy == actor.ID {
_ = s.Audit.Record(ctx, &actor.ID, "alert.ack", "alert", &alertID, "failure", map[string]any{"reason": "duplicate"})
return Result{Detail: detail, Idempotent: true}, nil
}
_ = s.Audit.Record(ctx, &actor.ID, "alert.ack", "alert", &alertID, "failure", map[string]any{"reason": "already_handled"})
return Result{Detail: detail}, ErrAlreadyHandled
}
func (s Service) Close(ctx context.Context, alertID string, input CloseInput, actor auth.User) (Result, error) {
if !validOutcome(input.Outcome) {
_ = s.Audit.Record(ctx, &actor.ID, "alert.close", "alert", &alertID, "failure", map[string]any{"reason": "outcome_required"})
return Result{}, ErrOutcomeRequired
}
current, err := s.Query.Get(ctx, alertID)
if err != nil {
return Result{}, err
}
if current.Alert.Status == "closed" {
if current.Alert.ClosedBy != nil && *current.Alert.ClosedBy == actor.ID && current.Alert.CloseOutcome != nil && *current.Alert.CloseOutcome == input.Outcome {
return Result{Detail: current, Idempotent: true}, nil
}
_ = s.Audit.Record(ctx, &actor.ID, "alert.close", "alert", &alertID, "failure", map[string]any{"reason": "already_closed"})
return Result{Detail: current}, ErrInvalidTransition
}
admin := hasRole(actor, "administrator")
if current.Alert.Status != "acknowledged" || (!admin && (current.Alert.AcknowledgedBy == nil || *current.Alert.AcknowledgedBy != actor.ID)) {
_ = s.Audit.Record(ctx, &actor.ID, "alert.close", "alert", &alertID, "failure", map[string]any{"reason": "invalid_owner_or_state"})
return Result{Detail: current}, ErrInvalidTransition
}
details, _ := json.Marshal(map[string]any{"outcome": input.Outcome, "note": input.Note})
tx, err := s.DB.Begin(ctx)
if err != nil {
return Result{}, err
}
defer func() { _ = tx.Rollback(ctx) }()
result, err := tx.Exec(ctx, `UPDATE bell_alerts SET status='closed',closed_by=$2,closed_at=now(),close_outcome=$3,close_note=nullif($4,''),updated_at=now() WHERE id=$1 AND status='acknowledged'`, alertID, actor.ID, input.Outcome, input.Note)
if err != nil {
return Result{}, err
}
if result.RowsAffected() != 1 {
return Result{}, ErrInvalidTransition
}
if _, err = tx.Exec(ctx, `INSERT INTO bell_alert_lifecycle_facts(alert_id,transition,actor_user_id,details) VALUES($1,'closed',$2,$3::jsonb)`, alertID, actor.ID, string(details)); err != nil {
return Result{}, err
}
if _, err = tx.Exec(ctx, `INSERT INTO bell_audit_log(actor_user_id,action,target_type,target_id,outcome,details) VALUES($1,'alert.close','alert',$2,'success',$3::jsonb)`, actor.ID, alertID, string(details)); err != nil {
return Result{}, err
}
if err = tx.Commit(ctx); err != nil {
return Result{}, err
}
detail, err := s.Query.Get(ctx, alertID)
return Result{Detail: detail}, err
}
@@ -0,0 +1,135 @@
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)
}
}
+30 -13
View File
@@ -3,16 +3,24 @@ package model
import "encoding/json"
type Alert struct {
ID string `json:"id"`
Status string `json:"status"`
Severity string `json:"severity"`
Summary string `json:"summary"`
Location string `json:"location"`
PrimaryRuleID string `json:"primary_rule_id"`
RuleName string `json:"rule_name"`
EventCount int `json:"event_count"`
CreatedAt string `json:"created_at"`
UpdatedAt string `json:"updated_at"`
ID string `json:"id"`
Status string `json:"status"`
Severity string `json:"severity"`
Summary string `json:"summary"`
Location string `json:"location"`
PrimaryRuleID string `json:"primary_rule_id"`
RuleName string `json:"rule_name"`
EventCount int `json:"event_count"`
AcknowledgedBy *string `json:"acknowledged_by,omitempty"`
AcknowledgedByName *string `json:"acknowledged_by_name,omitempty"`
AcknowledgedAt *string `json:"acknowledged_at,omitempty"`
ClosedBy *string `json:"closed_by,omitempty"`
ClosedByName *string `json:"closed_by_name,omitempty"`
ClosedAt *string `json:"closed_at,omitempty"`
CloseOutcome *string `json:"close_outcome,omitempty"`
CloseNote *string `json:"close_note,omitempty"`
CreatedAt string `json:"created_at"`
UpdatedAt string `json:"updated_at"`
}
type LinkedEvent struct {
ID string `json:"id"`
@@ -30,7 +38,16 @@ type Match struct {
MatchedAt string `json:"matched_at"`
}
type Detail struct {
Alert Alert `json:"alert"`
Events []LinkedEvent `json:"events"`
Matches []Match `json:"matches"`
Alert Alert `json:"alert"`
Events []LinkedEvent `json:"events"`
Matches []Match `json:"matches"`
Timeline []TimelineEntry `json:"timeline"`
}
type TimelineEntry struct {
ID int64 `json:"id"`
Transition string `json:"transition"`
ActorUserID string `json:"actor_user_id"`
ActorName string `json:"actor_name"`
OccurredAt string `json:"occurred_at"`
Details json.RawMessage `json:"details"`
}
+19 -3
View File
@@ -9,11 +9,11 @@ import (
type Service struct{ DB *pgxpool.Pool }
const base = `SELECT a.id::text,a.status,a.severity,a.summary,a.location,a.primary_rule_id::text,r.name,(SELECT count(*) FROM bell_alert_events ae WHERE ae.alert_id=a.id),a.created_at::text,a.updated_at::text FROM bell_alerts a JOIN bell_rules r ON r.id=a.primary_rule_id`
const base = `SELECT a.id::text,a.status,a.severity,a.summary,a.location,a.primary_rule_id::text,r.name,(SELECT count(*) FROM bell_alert_events ae WHERE ae.alert_id=a.id),a.acknowledged_by::text,ack_user.display_name,a.acknowledged_at::text,a.closed_by::text,close_user.display_name,a.closed_at::text,a.close_outcome,a.close_note,a.created_at::text,a.updated_at::text FROM bell_alerts a JOIN bell_rules r ON r.id=a.primary_rule_id LEFT JOIN bell_users ack_user ON ack_user.id=a.acknowledged_by LEFT JOIN bell_users close_user ON close_user.id=a.closed_by`
func scanAlert(row interface{ Scan(...any) error }) (model.Alert, error) {
var item model.Alert
err := row.Scan(&item.ID, &item.Status, &item.Severity, &item.Summary, &item.Location, &item.PrimaryRuleID, &item.RuleName, &item.EventCount, &item.CreatedAt, &item.UpdatedAt)
err := row.Scan(&item.ID, &item.Status, &item.Severity, &item.Summary, &item.Location, &item.PrimaryRuleID, &item.RuleName, &item.EventCount, &item.AcknowledgedBy, &item.AcknowledgedByName, &item.AcknowledgedAt, &item.ClosedBy, &item.ClosedByName, &item.ClosedAt, &item.CloseOutcome, &item.CloseNote, &item.CreatedAt, &item.UpdatedAt)
return item, err
}
func (s Service) List(ctx context.Context, limit int, before string) ([]model.Alert, error) {
@@ -40,7 +40,7 @@ func (s Service) Get(ctx context.Context, id string) (model.Detail, error) {
if err != nil {
return model.Detail{}, err
}
detail := model.Detail{Alert: item, Events: []model.LinkedEvent{}, Matches: []model.Match{}}
detail := model.Detail{Alert: item, Events: []model.LinkedEvent{}, Matches: []model.Match{}, Timeline: []model.TimelineEntry{}}
rows, err := s.DB.Query(ctx, `SELECT e.id::text,e.event_type,e.occurred_at::text,e.location,e.severity FROM bell_alert_events ae JOIN bell_events e ON e.id=ae.event_id WHERE ae.alert_id=$1 ORDER BY e.occurred_at,e.id`, id)
if err != nil {
return model.Detail{}, err
@@ -66,6 +66,22 @@ func (s Service) Get(ctx context.Context, id string) (model.Detail, error) {
}
detail.Matches = append(detail.Matches, match)
}
if err := rows.Err(); err != nil {
return model.Detail{}, err
}
rows.Close()
rows, err = s.DB.Query(ctx, `SELECT f.id,f.transition,f.actor_user_id::text,u.display_name,f.occurred_at::text,f.details FROM bell_alert_lifecycle_facts f JOIN bell_users u ON u.id=f.actor_user_id WHERE f.alert_id=$1 ORDER BY f.id`, id)
if err != nil {
return model.Detail{}, err
}
defer rows.Close()
for rows.Next() {
var entry model.TimelineEntry
if err := rows.Scan(&entry.ID, &entry.Transition, &entry.ActorUserID, &entry.ActorName, &entry.OccurredAt, &entry.Details); err != nil {
return model.Detail{}, err
}
detail.Timeline = append(detail.Timeline, entry)
}
return detail, rows.Err()
}
func (s Service) ForEvent(ctx context.Context, eventID string) ([]model.Alert, error) {
+4 -1
View File
@@ -12,6 +12,7 @@ import (
"syscall"
"time"
alertLifecycle "git.ilapage.cn/ila/yovision/Bell/server/app/alert/lifecycle"
alertQuery "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"
@@ -73,7 +74,9 @@ func Run(ctx context.Context, args []string) error {
eventService := event.Service{DB: db, AfterPersist: ruleService.Match}
event.HTTP{Service: eventService, Auth: authHTTP}.Register(app.Router())
rule.HTTP{Service: ruleService, Auth: authHTTP, Audit: auditStore}.Register(app.Router())
alertQuery.HTTP{Service: alertQuery.Service{DB: db}, Auth: authHTTP}.Register(app.Router())
alertQueryService := alertQuery.Service{DB: db}
alertQuery.HTTP{Service: alertQueryService, Auth: authHTTP}.Register(app.Router())
alertLifecycle.HTTP{Service: alertLifecycle.Service{DB: db, Audit: auditStore, Query: alertQueryService}, Auth: authHTTP}.Register(app.Router())
if cfg.Environment == "development" || cfg.Environment == "test" {
synthetic.HTTP{Service: synthetic.Service{Events: eventService}, Auth: authHTTP}.Register(app.Router())
}
@@ -0,0 +1,18 @@
ALTER TABLE bell_alerts
ADD COLUMN acknowledged_by uuid REFERENCES bell_users(id),
ADD COLUMN acknowledged_at timestamptz,
ADD COLUMN closed_by uuid REFERENCES bell_users(id),
ADD COLUMN closed_at timestamptz,
ADD COLUMN close_outcome text CHECK (close_outcome IN ('danger_confirmed','false_positive','site_normal','unable_to_confirm')),
ADD COLUMN close_note text;
CREATE TABLE bell_alert_lifecycle_facts (
id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
alert_id uuid NOT NULL REFERENCES bell_alerts(id),
transition text NOT NULL CHECK (transition IN ('acknowledged','closed')),
actor_user_id uuid NOT NULL REFERENCES bell_users(id),
occurred_at timestamptz NOT NULL DEFAULT now(),
details jsonb NOT NULL DEFAULT '{}'::jsonb,
UNIQUE(alert_id,transition)
);
CREATE INDEX bell_alert_lifecycle_timeline_idx ON bell_alert_lifecycle_facts(alert_id,id);
CREATE TRIGGER bell_alert_lifecycle_no_update BEFORE UPDATE OR DELETE ON bell_alert_lifecycle_facts FOR EACH ROW EXECUTE FUNCTION bell_reject_audit_mutation();