yovision/Bell/internal/store/event_ingress_postgres_test.go

package store

import (
	"context"
	"crypto/sha256"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"sync"
	"testing"
	"time"

	_ "github.com/jackc/pgx/v5/stdlib"

	"yovision/bell/contracts"
	"yovision/bell/internal/event"
	"yovision/bell/internal/ingress"
)

type ingressStoreIDs struct{ id string }

func (value ingressStoreIDs) NewEventID() (string, error) { return value.id, nil }

func ingressStoreCandidate(t *testing.T, sourceEventID string) []byte {
	t.Helper()
	var object map[string]any
	if err := json.Unmarshal(testCandidate(t, "brain-demo-v1"), &object); err != nil {
		t.Fatal(err)
	}
	object["source_event_id"] = sourceEventID
	object["tenant_id"] = float64(101)
	object["site_id"] = float64(201)
	object["device_id"] = float64(301)
	object["sensors"] = []any{map[string]any{"device_id": float64(301), "modality": "video", "role": "primary"}}
	object["kind"] = "zone_entry"
	object["severity"] = "medium"
	object["evidence"] = map[string]any{"snapshot_uris": []any{}, "clip_uri": nil, "clip_range": nil}
	encoded, err := json.Marshal(object)
	if err != nil {
		t.Fatal(err)
	}
	return encoded
}

func ingressStoreEvent(t *testing.T, repository *Postgres, id, sourceEventID string) event.Event {
	t.Helper()
	guard, err := event.NewEvidenceGuard("private-customer")
	if err != nil {
		t.Fatal(err)
	}
	factory, err := event.NewFactory(contracts.EventV01Schema, ingressStoreIDs{id}, repository, guard)
	if err != nil {
		t.Fatal(err)
	}
	value, err := factory.Create(context.Background(), ingressStoreCandidate(t, sourceEventID))
	if err != nil {
		t.Fatal(err)
	}
	return value
}

func TestPostgresEventIngressAtomicSourceReceipt(t *testing.T) {
	dsn := os.Getenv("YOVISION_TEST_BELL_POSTGRES_DSN")
	adminDSN := os.Getenv("YOVISION_TEST_POSTGRES_ADMIN_DSN")
	if dsn == "" || adminDSN == "" {
		t.Skip("Bell runtime and admin PostgreSQL DSNs are not set")
	}
	admin, err := sql.Open("pgx", adminDSN)
	if err != nil {
		t.Fatal(err)
	}
	defer admin.Close()
	ctx := context.Background()
	now := time.Now().UTC()
	statements := []struct {
		query string
		args  []any
	}{
		{`INSERT INTO bell.sites(tenant_id,id,name) VALUES ('ingress-tenant','ingress-site','Ingress Site')`, nil},
		{`INSERT INTO bell.areas(tenant_id,site_id,id,name,capture_policy) VALUES ('ingress-tenant','ingress-site','ingress-area','Ingress Area','video_allowed')`, nil},
		{`INSERT INTO sense.devices(id,tenant_id,site_id,serial_number,name,modality,desired_state,actual_state,area_id,created_at,updated_at)
		 VALUES ('ingress-device','ingress-tenant','ingress-site','INGRESS-SERIAL','Ingress Device','video','enabled','online','ingress-area',$1,$1)`, []any{now}},
		{`INSERT INTO bell.event_ingress_bindings(
		 producer_id,tenant_id,site_id,device_id,logical_tenant_id,logical_site_id,logical_device_id,logical_area_id,modality
		 ) VALUES ('brain-main',101,201,301,'ingress-tenant','ingress-site','ingress-device','ingress-area','video')`, nil},
	}
	for _, statement := range statements {
		if _, err := admin.ExecContext(ctx, statement.query, statement.args...); err != nil {
			t.Fatal(err)
		}
	}

	db, err := sql.Open("pgx", dsn)
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	repository, err := OpenPostgres(ctx, db)
	if err != nil {
		t.Fatal(err)
	}
	if err := repository.EventIngressReady(ctx); err != nil {
		t.Fatal(err)
	}
	candidate := ingressStoreCandidate(t, "BRN-ingress-0001")
	candidateHash := sha256.Sum256(candidate)
	value := ingressStoreEvent(t, repository, "evt_01J8XQ2K7M3P5R9T0V4W6Y8Z2C", "BRN-ingress-0001")
	requestHash := sha256.Sum256([]byte("first-request"))
	result, err := repository.ProcessEvent(ctx, "brain-a", "AAAAAAAAAAAAAAAAAAAAAA", requestHash, "brain-main", candidateHash, value)
	if err != nil || result.Status != "accepted" || result.HTTPStatus != 201 {
		t.Fatalf("first ingress: %+v %v", result, err)
	}

	// This is the crash-after-Bell-commit case: Brain uses a new nonce before
	// it has locally recorded the first response.
	replayed, found, err := repository.Replay(
		ctx, "brain-a", "BBBBBBBBBBBBBBBBBBBBBB", sha256.Sum256([]byte("retry-request")),
		"brain-main", "BRN-ingress-0001", candidateHash,
	)
	if err != nil || !found || replayed.Status != "duplicate" || replayed.EventID != result.EventID {
		t.Fatalf("durable source replay: %+v found=%v err=%v", replayed, found, err)
	}
	conflictHash := sha256.Sum256([]byte("changed-candidate"))
	if _, _, err := repository.Replay(
		ctx, "brain-a", "CCCCCCCCCCCCCCCCCCCCCC", sha256.Sum256([]byte("conflict-request")),
		"brain-main", "BRN-ingress-0001", conflictHash,
	); !errors.Is(err, ingress.ErrSourceConflict) {
		t.Fatalf("expected source conflict, got %v", err)
	}
	if _, _, err := repository.Replay(
		ctx, "brain-a", "AAAAAAAAAAAAAAAAAAAAAA", sha256.Sum256([]byte("different-request")),
		"brain-main", "BRN-ingress-0001", candidateHash,
	); !errors.Is(err, ingress.ErrReplayConflict) {
		t.Fatalf("expected nonce replay conflict, got %v", err)
	}

	var wait sync.WaitGroup
	errorsSeen := make(chan error, 8)
	concurrentValues := make([]event.Event, 8)
	for index := range concurrentValues {
		id := fmt.Sprintf("evt_01J8XQ2K7M3P5R9T0V4W6Y8Z2%c", "DEFGHJKM"[index])
		concurrentValues[index] = ingressStoreEvent(t, repository, id, "BRN-ingress-0001")
	}
	for index := 0; index < 8; index++ {
		wait.Add(1)
		go func(index int) {
			defer wait.Done()
			nonce := base64Nonce(index)
			response, err := repository.ProcessEvent(
				ctx, "brain-a", nonce, sha256.Sum256([]byte(nonce)), "brain-main", candidateHash, concurrentValues[index],
			)
			if err != nil || response.Status != "duplicate" || response.EventID != result.EventID {
				errorsSeen <- fmt.Errorf("concurrent duplicate %d: %+v %w", index, response, err)
			}
		}(index)
	}
	wait.Wait()
	close(errorsSeen)
	for err := range errorsSeen {
		t.Error(err)
	}
	var eventCount, receiptCount int
	if err := db.QueryRowContext(ctx, `SELECT
		(SELECT count(*) FROM bell.events WHERE tenant_id=101 AND site_id=201 AND source_event_id='BRN-ingress-0001'),
		(SELECT count(*) FROM bell.event_ingress_receipts WHERE producer_id='brain-main' AND source_event_id='BRN-ingress-0001')`).Scan(&eventCount, &receiptCount); err != nil {
		t.Fatal(err)
	}
	if eventCount != 1 || receiptCount != 1 {
		t.Fatalf("concurrent ingress created events=%d receipts=%d", eventCount, receiptCount)
	}

	if _, err := admin.ExecContext(ctx, `UPDATE bell.areas SET capture_policy='non_imaging_only'
		WHERE tenant_id='ingress-tenant' AND id='ingress-area'`); err != nil {
		t.Fatal(err)
	}
	allowed, err := repository.VideoAllowed(ctx, 101, 201, 301)
	if err != nil || allowed {
		t.Fatalf("privacy change did not fail closed: allowed=%v err=%v", allowed, err)
	}
	guard, _ := event.NewEvidenceGuard("private-customer")
	factory, _ := event.NewFactory(contracts.EventV01Schema, ingressStoreIDs{"evt_01J8XQ2K7M3P5R9T0V4W6Y8Z2N"}, repository, guard)
	_, err = factory.Create(ctx, ingressStoreCandidate(t, "BRN-ingress-0002"))
	var validation *event.ValidationError
	if !errors.As(err, &validation) || validation.Code != event.CodePrivacyDenied {
		t.Fatalf("privacy denial code drift: %v", err)
	}
	if _, err := db.ExecContext(ctx, `UPDATE bell.event_ingress_receipts SET event_id=event_id
		WHERE producer_id='brain-main' AND source_event_id='BRN-ingress-0001'`); err == nil {
		t.Fatal("runtime updated immutable event source receipt")
	}
}

func base64Nonce(index int) string {
	return fmt.Sprintf("D%021d", index)
}