yovision/Bell/server/app/bell/integration/event_ingress/service.go

package event_ingress

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"time"

	"github.com/google/uuid"
	"gorm.io/gorm"
	"gorm.io/gorm/clause"

	"go-admin/app/bell/event"
	"go-admin/app/bell/receipt"
)

var ErrIdempotencyConflict = errors.New("idempotency_conflict")

type EvidenceRefresher interface {
	Refresh(context.Context, *gorm.DB, EvidenceStatus) error
}

type Service struct {
	DB       *gorm.DB
	Resolver EvidenceRefresher
}

func (s Service) Ingest(ctx context.Context, parsed ParsedEvent) (IngestResult, error) {
	if s.DB == nil {
		return IngestResult{}, errors.New("event database is unavailable")
	}
	var output IngestResult
	err := s.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
		if tx.Dialector.Name() == "postgres" {
			key := fmt.Sprintf("%d:%s:%s", len(parsed.Event.ProducerID), parsed.Event.ProducerID, parsed.Event.SourceEventID)
			if err := tx.Exec("SELECT pg_advisory_xact_lock(hashtextextended(?, 0))", key).Error; err != nil {
				return err
			}
		}
		var existing struct {
			EventID       string
			PayloadSHA256 string
		}
		err := tx.Model(&receipt.Receipt{}).Clauses(clause.Locking{Strength: "UPDATE"}).
			Select("event_id", "payload_sha256").Where("producer_id = ? AND source_event_id = ?", parsed.Event.ProducerID, parsed.Event.SourceEventID).First(&existing).Error
		if err == nil {
			if existing.PayloadSHA256 != parsed.Digest {
				output.EventID = existing.EventID
				return ErrIdempotencyConflict
			}
			output = IngestResult{EventID: existing.EventID, ProducerID: parsed.Event.ProducerID, SourceEventID: parsed.Event.SourceEventID, Disposition: "duplicate", PayloadSHA256: parsed.Digest}
			return tx.Create(&receipt.IngestAudit{ProducerID: parsed.Event.ProducerID, SourceEventID: parsed.Event.SourceEventID, PayloadSHA256: parsed.Digest, Outcome: receipt.OutcomeReplay, ActorID: 0, CreatedAt: time.Now().UTC()}).Error
		}
		if !errors.Is(err, gorm.ErrRecordNotFound) {
			return err
		}
		now := time.Now().UTC()
		eventID := uuid.NewString()
		var evidenceRef *string
		if len(parsed.Event.Evidence) > 0 {
			value := parsed.Event.Evidence[0].EvidenceID
			evidenceRef = &value
		}
		item := event.Event{ID: eventID, ProducerID: parsed.Event.ProducerID, SourceEventID: parsed.Event.SourceEventID,
			EventType: parsed.Event.EventType, OccurredAt: parsed.Occurred, Location: parsed.Event.SiteRef + "/" + parsed.Event.DeviceRef,
			Severity: parsed.Event.Severity, EvidenceRef: evidenceRef, NormalizedPayload: parsed.Canonical, PayloadSHA256: parsed.Digest, ReceivedAt: now}
		receiptItem := receipt.Receipt{ID: uuid.NewString(), EventID: eventID, ProducerID: parsed.Event.ProducerID, SourceEventID: parsed.Event.SourceEventID, PayloadSHA256: parsed.Digest, AcceptedAt: now}
		if err := tx.Create(&item).Error; err != nil {
			return err
		}
		if err := tx.Create(&receiptItem).Error; err != nil {
			return err
		}
		for _, evidence := range parsed.Event.Evidence {
			payload, marshalErr := json.Marshal(evidence)
			if marshalErr != nil {
				return marshalErr
			}
			canonical, canonicalErr := canonicalJSON(payload)
			if canonicalErr != nil {
				return canonicalErr
			}
			var expiresAt *time.Time
			if evidence.ExpiresAt != "" {
				parsedExpiry, parseErr := time.Parse(time.RFC3339Nano, evidence.ExpiresAt)
				if parseErr != nil {
					return parseErr
				}
				parsedExpiry = parsedExpiry.UTC()
				expiresAt = &parsedExpiry
			}
			status := EvidenceStatus{EventID: eventID, EvidenceID: evidence.EvidenceID, OwnerID: evidence.OwnerID,
				Status: evidence.Status, Resolution: "snapshot", CurrentPayload: canonical, ExpiresAt: expiresAt, CreatedAt: now, UpdatedAt: now}
			if err := tx.Create(&status).Error; err != nil {
				return err
			}
		}
		if err := tx.Create(&receipt.IngestAudit{ProducerID: parsed.Event.ProducerID, SourceEventID: parsed.Event.SourceEventID, PayloadSHA256: parsed.Digest, Outcome: receipt.OutcomeAccepted, ActorID: 0, CreatedAt: now}).Error; err != nil {
			return err
		}
		output = IngestResult{EventID: eventID, ProducerID: parsed.Event.ProducerID, SourceEventID: parsed.Event.SourceEventID, Disposition: "created", PayloadSHA256: parsed.Digest}
		return nil
	})
	if errors.Is(err, ErrIdempotencyConflict) {
		auditErr := s.DB.WithContext(ctx).Create(&receipt.IngestAudit{ProducerID: parsed.Event.ProducerID, SourceEventID: parsed.Event.SourceEventID, PayloadSHA256: parsed.Digest, Outcome: receipt.OutcomeConflict, ActorID: 0, CreatedAt: time.Now().UTC()}).Error
		if auditErr != nil {
			return IngestResult{}, fmt.Errorf("record conflict audit: %w", auditErr)
		}
		return output, ErrIdempotencyConflict
	}
	if err != nil || s.Resolver == nil {
		return output, err
	}
	var statuses []EvidenceStatus
	if err = s.DB.WithContext(ctx).Where("event_id = ?", output.EventID).Find(&statuses).Error; err != nil {
		return IngestResult{}, err
	}
	for _, status := range statuses {
		// Evidence lookup is supplementary. The immutable Event/Receipt boundary
		// remains accepted even when Sense is unavailable.
		_ = s.Resolver.Refresh(ctx, s.DB.WithContext(ctx), status)
	}
	return output, nil
}