yovision/Bell/server/app/event/service.go

package event

import (
	"context"
	"crypto/sha256"
	"encoding/json"
	"errors"
	"fmt"
	"strings"

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

var ErrConflict = errors.New("幂等键已用于不同事件载荷")
var ErrInvalid = errors.New("事件字段无效")

type PersistHook func(context.Context, pgx.Tx, Event) error
type Service struct {
	DB           *pgxpool.Pool
	AfterPersist PersistHook
}

func normalize(command Command) ([]byte, [32]byte, error) {
	command.ProducerID = strings.TrimSpace(command.ProducerID)
	command.SourceEventID = strings.TrimSpace(command.SourceEventID)
	command.EventType = strings.TrimSpace(command.EventType)
	command.Location = strings.TrimSpace(command.Location)
	command.Severity = strings.ToLower(strings.TrimSpace(command.Severity))
	command.OccurredAt = command.OccurredAt.UTC()
	if command.ProducerID == "" || command.SourceEventID == "" || command.EventType == "" || command.Location == "" || command.OccurredAt.IsZero() {
		return nil, [32]byte{}, ErrInvalid
	}
	if command.Severity != "low" && command.Severity != "medium" && command.Severity != "high" && command.Severity != "critical" {
		return nil, [32]byte{}, ErrInvalid
	}
	if command.EvidenceRef != nil {
		value := strings.TrimSpace(*command.EvidenceRef)
		if value == "" || strings.Contains(value, "@") || strings.Contains(value, "\\") || strings.HasPrefix(strings.ToLower(value), "file:") {
			return nil, [32]byte{}, fmt.Errorf("%w: evidence_ref must be a safe logical reference", ErrInvalid)
		}
		command.EvidenceRef = &value
	}
	if command.Attributes == nil {
		command.Attributes = map[string]any{}
	}
	data, err := json.Marshal(command)
	if err != nil {
		return nil, [32]byte{}, fmt.Errorf("normalize event: %w", err)
	}
	return data, sha256.Sum256(data), nil
}

func (s Service) Ingest(ctx context.Context, command Command) (Result, error) {
	payload, digest, err := normalize(command)
	if err != nil {
		return Result{}, err
	}
	var normalized Command
	if err := json.Unmarshal(payload, &normalized); err != nil {
		return Result{}, err
	}
	tx, err := s.DB.Begin(ctx)
	if err != nil {
		return Result{}, err
	}
	defer func() { _ = tx.Rollback(ctx) }()
	var eventID string
	err = tx.QueryRow(ctx, `INSERT INTO bell_events(producer_id,source_event_id,event_type,occurred_at,location,severity,evidence_ref,normalized_payload,payload_sha256) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9) ON CONFLICT(producer_id,source_event_id) DO NOTHING RETURNING id::text`, normalized.ProducerID, normalized.SourceEventID, normalized.EventType, normalized.OccurredAt, normalized.Location, normalized.Severity, normalized.EvidenceRef, payload, digest[:]).Scan(&eventID)
	created := err == nil
	if err != nil && !errors.Is(err, pgx.ErrNoRows) {
		return Result{}, err
	}
	if created {
		if _, err = tx.Exec(ctx, `INSERT INTO bell_event_receipts(event_id,producer_id,source_event_id,payload_sha256) VALUES($1,$2,$3,$4)`, eventID, normalized.ProducerID, normalized.SourceEventID, digest[:]); err != nil {
			return Result{}, err
		}
	}
	result, storedDigest, err := loadResult(ctx, tx, normalized.ProducerID, normalized.SourceEventID)
	if err != nil {
		return Result{}, err
	}
	if !created && !equalDigest(storedDigest, digest[:]) {
		return Result{}, ErrConflict
	}
	result.Duplicate = !created
	if created && s.AfterPersist != nil {
		if err := s.AfterPersist(ctx, tx, result.Event); err != nil {
			return Result{}, fmt.Errorf("process accepted Event: %w", err)
		}
	}
	if err = tx.Commit(ctx); err != nil {
		return Result{}, err
	}
	return result, nil
}

func loadResult(ctx context.Context, q interface {
	QueryRow(context.Context, string, ...any) pgx.Row
}, producerID, sourceEventID string) (Result, []byte, error) {
	var result Result
	var digest []byte
	err := q.QueryRow(ctx, `SELECT e.id::text,e.producer_id,e.source_event_id,e.event_type,e.occurred_at::text,e.location,e.severity,e.evidence_ref,e.normalized_payload,e.created_at::text,e.payload_sha256,r.id::text,r.accepted_at::text FROM bell_events e JOIN bell_event_receipts r ON r.event_id=e.id WHERE e.producer_id=$1 AND e.source_event_id=$2`, producerID, sourceEventID).Scan(&result.Event.ID, &result.Event.ProducerID, &result.Event.SourceEventID, &result.Event.EventType, &result.Event.OccurredAt, &result.Event.Location, &result.Event.Severity, &result.Event.EvidenceRef, &result.Event.NormalizedPayload, &result.Event.CreatedAt, &digest, &result.Receipt.ID, &result.Receipt.AcceptedAt)
	result.Receipt.EventID = result.Event.ID
	result.Receipt.ProducerID = producerID
	result.Receipt.SourceEventID = sourceEventID
	return result, digest, err
}
func equalDigest(a, b []byte) bool {
	if len(a) != len(b) {
		return false
	}
	var different byte
	for i := range a {
		different |= a[i] ^ b[i]
	}
	return different == 0
}

func (s Service) List(ctx context.Context, limit int, before string) ([]Event, error) {
	if limit < 1 || limit > 100 {
		limit = 25
	}
	rows, err := s.DB.Query(ctx, `SELECT id::text,producer_id,source_event_id,event_type,occurred_at::text,location,severity,evidence_ref,normalized_payload,created_at::text FROM bell_events WHERE ($2='' OR (created_at,id)<((SELECT created_at FROM bell_events WHERE id=nullif($2,'')::uuid),nullif($2,'')::uuid)) ORDER BY created_at DESC,id DESC LIMIT $1`, limit, before)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	items := []Event{}
	for rows.Next() {
		var item Event
		if err := rows.Scan(&item.ID, &item.ProducerID, &item.SourceEventID, &item.EventType, &item.OccurredAt, &item.Location, &item.Severity, &item.EvidenceRef, &item.NormalizedPayload, &item.CreatedAt); err != nil {
			return nil, err
		}
		items = append(items, item)
	}
	return items, rows.Err()
}
func (s Service) Get(ctx context.Context, id string) (Event, error) {
	var item Event
	err := s.DB.QueryRow(ctx, `SELECT id::text,producer_id,source_event_id,event_type,occurred_at::text,location,severity,evidence_ref,normalized_payload,created_at::text FROM bell_events WHERE id=$1`, id).Scan(&item.ID, &item.ProducerID, &item.SourceEventID, &item.EventType, &item.OccurredAt, &item.Location, &item.Severity, &item.EvidenceRef, &item.NormalizedPayload, &item.CreatedAt)
	return item, err
}