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 }