111 lines
3.9 KiB
Go
111 lines
3.9 KiB
Go
package event
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"time"
|
|
|
|
"github.com/google/uuid"
|
|
"gorm.io/gorm"
|
|
|
|
"go-admin/app/bell/receipt"
|
|
)
|
|
|
|
type Result struct {
|
|
Event Event `json:"event"`
|
|
Receipt receipt.Receipt `json:"receipt"`
|
|
Duplicate bool `json:"duplicate"`
|
|
}
|
|
|
|
type Service struct{ DB *gorm.DB }
|
|
|
|
func NewService(db *gorm.DB) Service { return Service{DB: db} }
|
|
|
|
func (s Service) Ingest(ctx context.Context, command Command, actorID int) (Result, error) {
|
|
if s.DB == nil {
|
|
return Result{}, errors.New("event database is unavailable")
|
|
}
|
|
normalized, err := Normalize(command)
|
|
if err != nil {
|
|
return Result{}, err
|
|
}
|
|
var result Result
|
|
err = s.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
|
lockKey := fmt.Sprintf("%d:%s:%s", len(normalized.Command.ProducerID), normalized.Command.ProducerID, normalized.Command.SourceEventID)
|
|
if err := tx.Exec("SELECT pg_advisory_xact_lock(hashtextextended(?, 0))", lockKey).Error; err != nil {
|
|
return fmt.Errorf("lock event idempotency key: %w", err)
|
|
}
|
|
|
|
var existing receipt.Receipt
|
|
err := tx.Where("producer_id = ? AND source_event_id = ?", normalized.Command.ProducerID, normalized.Command.SourceEventID).
|
|
First(&existing).Error
|
|
if err == nil {
|
|
if existing.PayloadSHA256 != normalized.Digest {
|
|
return ErrConflict
|
|
}
|
|
if err = tx.First(&result.Event, "id = ?", existing.EventID).Error; err != nil {
|
|
return fmt.Errorf("load replay event: %w", err)
|
|
}
|
|
result.Receipt = existing
|
|
result.Duplicate = true
|
|
return tx.Create(&receipt.IngestAudit{
|
|
ProducerID: normalized.Command.ProducerID, SourceEventID: normalized.Command.SourceEventID,
|
|
PayloadSHA256: normalized.Digest, Outcome: receipt.OutcomeReplay, ActorID: actorID, CreatedAt: time.Now().UTC(),
|
|
}).Error
|
|
}
|
|
if !errors.Is(err, gorm.ErrRecordNotFound) {
|
|
return fmt.Errorf("load event receipt: %w", err)
|
|
}
|
|
|
|
now := time.Now().UTC()
|
|
result.Event = Event{
|
|
ID: uuid.NewString(), ProducerID: normalized.Command.ProducerID,
|
|
SourceEventID: normalized.Command.SourceEventID, EventType: normalized.Command.EventType,
|
|
OccurredAt: normalized.Command.OccurredAt, Location: normalized.Command.Location,
|
|
Severity: normalized.Command.Severity, EvidenceRef: normalized.Command.EvidenceRef,
|
|
NormalizedPayload: normalized.Payload, PayloadSHA256: normalized.Digest, ReceivedAt: now,
|
|
}
|
|
result.Receipt = receipt.Receipt{
|
|
ID: uuid.NewString(), EventID: result.Event.ID, ProducerID: result.Event.ProducerID,
|
|
SourceEventID: result.Event.SourceEventID, PayloadSHA256: normalized.Digest, AcceptedAt: now,
|
|
}
|
|
if err = tx.Create(&result.Event).Error; err != nil {
|
|
return fmt.Errorf("create event: %w", err)
|
|
}
|
|
if err = tx.Create(&result.Receipt).Error; err != nil {
|
|
return fmt.Errorf("create event receipt: %w", err)
|
|
}
|
|
return tx.Create(&receipt.IngestAudit{
|
|
ProducerID: result.Event.ProducerID, SourceEventID: result.Event.SourceEventID,
|
|
PayloadSHA256: normalized.Digest, Outcome: receipt.OutcomeAccepted, ActorID: actorID, CreatedAt: now,
|
|
}).Error
|
|
})
|
|
if errors.Is(err, ErrConflict) {
|
|
// The conflict audit must commit independently from the rejected ingest.
|
|
auditErr := s.DB.WithContext(ctx).Create(&receipt.IngestAudit{
|
|
ProducerID: normalized.Command.ProducerID, SourceEventID: normalized.Command.SourceEventID,
|
|
PayloadSHA256: normalized.Digest, Outcome: receipt.OutcomeConflict, ActorID: actorID, CreatedAt: time.Now().UTC(),
|
|
}).Error
|
|
if auditErr != nil {
|
|
return Result{}, fmt.Errorf("record idempotency conflict audit: %w", auditErr)
|
|
}
|
|
return Result{}, ErrConflict
|
|
}
|
|
return result, err
|
|
}
|
|
|
|
func (s Service) Get(ctx context.Context, id string) (Event, error) {
|
|
if _, err := uuid.Parse(id); err != nil {
|
|
return Event{}, ErrNotFound
|
|
}
|
|
var item Event
|
|
if err := s.DB.WithContext(ctx).First(&item, "id = ?", id).Error; err != nil {
|
|
if errors.Is(err, gorm.ErrRecordNotFound) {
|
|
return Event{}, ErrNotFound
|
|
}
|
|
return Event{}, err
|
|
}
|
|
return item, nil
|
|
}
|