feat(bell): add immutable event store [T-015]

2026-08-10 23:53:11 +08:00
parent 5b42f58f23
commit 236964d242
28 changed files with 1434 additions and 27 deletions
@@ -0,0 +1,149 @@
+package store
+
+import (
+	"bytes"
+	"context"
+	"crypto/sha256"
+	"database/sql"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"regexp"
+	"time"
+	"unicode/utf8"
+
+	"yovision/bell/internal/event"
+)
+
+var ErrConflict = errors.New("immutable record id conflict")
+
+type Postgres struct {
+	db *sql.DB
+}
+
+func OpenPostgres(ctx context.Context, db *sql.DB) (*Postgres, error) {
+	if db == nil {
+		return nil, errors.New("postgres database is required")
+	}
+	if err := db.PingContext(ctx); err != nil {
+		return nil, fmt.Errorf("ping Bell postgres: %w", err)
+	}
+	var version int64
+	if err := db.QueryRowContext(ctx, `SELECT COALESCE(MAX(version), 0) FROM bell.schema_migrations`).Scan(&version); err != nil || version < 3 {
+		return nil, errors.New("postgres Bell schema migration v3 is required")
+	}
+	var canInsert, canSelect, canUpdate, canDelete, canTruncate bool
+	if err := db.QueryRowContext(ctx, `SELECT
+		has_table_privilege(current_user, 'bell.events', 'INSERT'),
+		has_table_privilege(current_user, 'bell.events', 'SELECT'),
+		has_table_privilege(current_user, 'bell.events', 'UPDATE'),
+		has_table_privilege(current_user, 'bell.events', 'DELETE'),
+		has_table_privilege(current_user, 'bell.events', 'TRUNCATE')`).Scan(
+		&canInsert, &canSelect, &canUpdate, &canDelete, &canTruncate,
+	); err != nil {
+		return nil, fmt.Errorf("inspect Bell event privileges: %w", err)
+	}
+	if !canInsert || !canSelect || canUpdate || canDelete || canTruncate {
+		return nil, errors.New("Bell runtime event privileges violate append-only boundary")
+	}
+	return &Postgres{db: db}, nil
+}
+
+// InsertEvent is idempotent only for the same platform ID and exact payload.
+func (p *Postgres) InsertEvent(ctx context.Context, value event.Event) (bool, error) {
+	digest := value.Digest()
+	result, err := p.db.ExecContext(ctx, `INSERT INTO bell.events(
+		id, tenant_id, site_id, device_id, source_event_id, kind, severity,
+		occurred_at, detected_at, payload_hash, payload
+	) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11::jsonb)
+	ON CONFLICT (id) DO NOTHING`,
+		value.ID(), value.TenantID(), value.SiteID(), value.DeviceID(),
+		value.SourceEventID(), value.Kind(), value.Severity(), value.OccurredAt(),
+		value.DetectedAt(), digest[:], value.JSON(),
+	)
+	if err != nil {
+		return false, fmt.Errorf("insert immutable Bell event: %w", err)
+	}
+	rows, err := result.RowsAffected()
+	if err != nil {
+		return false, fmt.Errorf("read Bell event insert result: %w", err)
+	}
+	if rows == 1 {
+		return true, nil
+	}
+	var existing []byte
+	if err := p.db.QueryRowContext(ctx, `SELECT payload_hash FROM bell.events WHERE id=$1`, value.ID()).Scan(&existing); err != nil {
+		return false, fmt.Errorf("read existing Bell event digest: %w", err)
+	}
+	if !bytes.Equal(existing, digest[:]) {
+		return false, ErrConflict
+	}
+	return false, nil
+}
+
+type Outcome struct {
+	ID         string    `json:"id"`
+	EventID    string    `json:"event_id"`
+	Value      string    `json:"outcome"`
+	Source     string    `json:"source"`
+	Reason     *string   `json:"reason"`
+	ActorType  string    `json:"actor_type"`
+	ActorID    string    `json:"actor_id"`
+	OccurredAt time.Time `json:"occurred_at"`
+}
+
+var outcomeID = regexp.MustCompile(`^out_[0-9A-HJKMNP-TV-Z]{26}$`)
+var eventID = regexp.MustCompile(`^evt_[0-9A-HJKMNP-TV-Z]{26}$`)
+
+func (o Outcome) validate() error {
+	if !outcomeID.MatchString(o.ID) || !eventID.MatchString(o.EventID) || o.ActorID == "" || o.OccurredAt.IsZero() {
+		return errors.New("invalid outcome identity")
+	}
+	validOutcome := map[string]bool{"unknown": true, "true_positive": true, "false_positive": true, "subject_recovered": true, "duplicate": true, "test": true}
+	if !validOutcome[o.Value] || (o.Source != "auto" && o.Source != "manual") {
+		return errors.New("invalid outcome value or source")
+	}
+	if o.ActorType != "user" && o.ActorType != "service" && o.ActorType != "system" {
+		return errors.New("invalid outcome actor type")
+	}
+	if o.Reason != nil && utf8.RuneCountInString(*o.Reason) > 500 {
+		return errors.New("outcome reason is too long")
+	}
+	return nil
+}
+
+// AppendOutcome never mutates the event or an earlier outcome record.
+func (p *Postgres) AppendOutcome(ctx context.Context, value Outcome) (bool, error) {
+	if err := value.validate(); err != nil {
+		return false, err
+	}
+	encoded, err := json.Marshal(value)
+	if err != nil {
+		return false, fmt.Errorf("encode outcome: %w", err)
+	}
+	digest := sha256.Sum256(encoded)
+	result, err := p.db.ExecContext(ctx, `INSERT INTO bell.event_outcomes(
+		id, event_id, outcome, outcome_source, reason, actor_type, actor_id,
+		occurred_at, record_hash
+	) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9)
+	ON CONFLICT (id) DO NOTHING`, value.ID, value.EventID, value.Value, value.Source,
+		value.Reason, value.ActorType, value.ActorID, value.OccurredAt, digest[:])
+	if err != nil {
+		return false, fmt.Errorf("append Bell event outcome: %w", err)
+	}
+	rows, err := result.RowsAffected()
+	if err != nil {
+		return false, fmt.Errorf("read Bell outcome insert result: %w", err)
+	}
+	if rows == 1 {
+		return true, nil
+	}
+	var existing []byte
+	if err := p.db.QueryRowContext(ctx, `SELECT record_hash FROM bell.event_outcomes WHERE id=$1`, value.ID).Scan(&existing); err != nil {
+		return false, fmt.Errorf("read existing Bell outcome digest: %w", err)
+	}
+	if !bytes.Equal(existing, digest[:]) {
+		return false, ErrConflict
+	}
+	return false, nil
+}
@@ -0,0 +1,120 @@
+package store
+
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+	"errors"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	_ "github.com/jackc/pgx/v5/stdlib"
+
+	"yovision/bell/contracts"
+	"yovision/bell/internal/event"
+)
+
+type storeIDs struct{ id string }
+
+func (f storeIDs) NewEventID() (string, error) { return f.id, nil }
+
+type allowVideo struct{}
+
+func (allowVideo) VideoAllowed(context.Context, int64, int64, int64) (bool, error) {
+	return true, nil
+}
+
+func testCandidate(t *testing.T, configVersion string) []byte {
+	t.Helper()
+	path := filepath.Join("..", "..", "..", "docs", "raw", "contracts", "event-v0.1.example-current.json")
+	raw, err := os.ReadFile(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var object map[string]any
+	if err := json.Unmarshal(raw, &object); err != nil {
+		t.Fatal(err)
+	}
+	delete(object, "id")
+	object["config_version"] = configVersion
+	encoded, err := json.Marshal(object)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return encoded
+}
+
+func newEvent(t *testing.T, configVersion string) event.Event {
+	t.Helper()
+	guard, err := event.NewEvidenceGuard()
+	if err != nil {
+		t.Fatal(err)
+	}
+	factory, err := event.NewFactory(
+		contracts.EventV01Schema,
+		storeIDs{id: "evt_01J8XQ2K7M3P5R9T0V4W6Y8Z2B"},
+		allowVideo{}, guard,
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+	value, err := factory.Create(context.Background(), testCandidate(t, configVersion))
+	if err != nil {
+		t.Fatal(err)
+	}
+	return value
+}
+
+func TestPostgresImmutableEventAndOutcome(t *testing.T) {
+	dsn := os.Getenv("YOVISION_TEST_BELL_POSTGRES_DSN")
+	if dsn == "" {
+		t.Skip("YOVISION_TEST_BELL_POSTGRES_DSN is not set")
+	}
+	db, err := sql.Open("pgx", dsn)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+	ctx := context.Background()
+	repo, err := OpenPostgres(ctx, db)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	value := newEvent(t, "sp-v1-2026.07.20")
+	created, err := repo.InsertEvent(ctx, value)
+	if err != nil || !created {
+		t.Fatalf("first insert: created=%v err=%v", created, err)
+	}
+	created, err = repo.InsertEvent(ctx, value)
+	if err != nil || created {
+		t.Fatalf("idempotent replay: created=%v err=%v", created, err)
+	}
+	if _, err := repo.InsertEvent(ctx, newEvent(t, "sp-v1-conflict")); !errors.Is(err, ErrConflict) {
+		t.Fatalf("expected immutable conflict, got %v", err)
+	}
+
+	reason := "confirmed by operator"
+	outcome := Outcome{
+		ID: "out_01J8XQ2K7M3P5R9T0V4W6Y8Z2B", EventID: value.ID(),
+		Value: "true_positive", Source: "manual", Reason: &reason,
+		ActorType: "user", ActorID: "operator-1", OccurredAt: time.Now().UTC(),
+	}
+	created, err = repo.AppendOutcome(ctx, outcome)
+	if err != nil || !created {
+		t.Fatalf("append outcome: created=%v err=%v", created, err)
+	}
+	created, err = repo.AppendOutcome(ctx, outcome)
+	if err != nil || created {
+		t.Fatalf("idempotent outcome replay: created=%v err=%v", created, err)
+	}
+
+	if _, err := db.ExecContext(ctx, `UPDATE bell.events SET kind='changed' WHERE id=$1`, value.ID()); err == nil {
+		t.Fatal("runtime unexpectedly updated immutable event")
+	}
+	if _, err := db.ExecContext(ctx, `DELETE FROM bell.event_outcomes WHERE id=$1`, outcome.ID); err == nil {
+		t.Fatal("runtime unexpectedly deleted immutable outcome")
+	}
+}