You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airavata.apache.org by la...@apache.org on 2026/09/25 16:03:14 UTC
(airavata-custos) 05/05: Publish events inside the service transaction
This is an automated email from the ASF dual-hosted git repository.
lahirujayathilake pushed a commit to branch persistent-event-bus
in repository https://gitbox.apache.org/repos/asf/airavata-custos.git
commit 615d76d40ae16786e8a0a487640554d05acf00e7
Author: lahiruj <la...@gmail.com>
AuthorDate: Fri Sep 25 01:35:47 2026 -0400
Publish events inside the service transaction
---
pkg/events/bus.go | 47 +++++++-----------
pkg/events/bus_test.go | 58 ++++++++++++++--------
pkg/service/compute_allocation.go | 21 ++++----
pkg/service/compute_allocation_diff.go | 14 +++---
pkg/service/compute_allocation_membership.go | 28 ++++++-----
...pute_allocation_membership_resource_override.go | 21 ++++----
pkg/service/compute_allocation_resource.go | 21 ++++----
pkg/service/compute_allocation_resource_mapping.go | 21 ++++----
pkg/service/compute_cluster.go | 21 ++++----
pkg/service/compute_cluster_user.go | 21 ++++----
pkg/service/identity_resolver.go | 10 ++--
pkg/service/organization.go | 21 ++++----
pkg/service/project.go | 30 ++++++-----
pkg/service/user.go | 30 ++++++-----
pkg/service/user_identity.go | 21 ++++----
pkg/service/user_merge.go | 13 +++--
pkg/service/user_onboard.go | 12 +++--
17 files changed, 231 insertions(+), 179 deletions(-)
diff --git a/pkg/events/bus.go b/pkg/events/bus.go
index b574eba63..34ef889e5 100644
--- a/pkg/events/bus.go
+++ b/pkg/events/bus.go
@@ -31,7 +31,6 @@ import (
"go.opentelemetry.io/otel/codes"
"github.com/apache/airavata-custos/internal/audit"
- "github.com/apache/airavata-custos/internal/db"
"github.com/apache/airavata-custos/internal/store"
"github.com/apache/airavata-custos/internal/tracing"
"github.com/apache/airavata-custos/pkg/models"
@@ -113,11 +112,10 @@ func subscribeTyped[T any](b *Bus, subscriber string, topic EventType, handler f
}
// Publish stores the event with one delivery per saved subscriber, and the
-// worker delivers it later. Nothing is written when nobody subscribes.
-func (b *Bus) Publish(ctx context.Context, topic EventType, payload any) error {
- ctx, span := tracing.Start(ctx, "bus.publish:"+string(topic))
- defer span.End()
-
+// worker delivers it later. It writes the event and its deliveries in the caller's
+// transaction, so everything is committed or rolled back together.
+// Nothing is written when nobody subscribes.
+func (b *Bus) Publish(ctx context.Context, tx *sql.Tx, topic EventType, payload any) error {
subscribers := b.subscribersOf(string(topic))
if len(subscribers) == 0 {
return nil
@@ -143,31 +141,22 @@ func (b *Bus) Publish(ctx context.Context, topic EventType, payload any) error {
CreatedAt: now,
}
- err = db.TxFn(ctx, b.db, func(tx *sql.Tx) error {
- if err := b.store.CreateEvent(ctx, tx, event); err != nil {
- return err
+ if err := b.store.CreateEvent(ctx, tx, event); err != nil {
+ return fmt.Errorf("store event: %w", err)
+ }
+ // Create a delivery for each subscriber with a PENDING status
+ for _, name := range subscribers {
+ d := &models.EventDelivery{
+ ID: uuid.NewString(),
+ EventID: event.ID,
+ Subscriber: name,
+ Status: models.EventDeliveryPending,
+ NextRunAt: now,
+ CreatedAt: now,
}
- // Create a delivery for each subscriber with a PENDING status
- for _, name := range subscribers {
- d := &models.EventDelivery{
- ID: uuid.NewString(),
- EventID: event.ID,
- Subscriber: name,
- Status: models.EventDeliveryPending,
- NextRunAt: now,
- CreatedAt: now,
- }
- if err := b.store.CreateDelivery(ctx, tx, d); err != nil {
- return err
- }
+ if err := b.store.CreateDelivery(ctx, tx, d); err != nil {
+ return fmt.Errorf("store event delivery: %w", err)
}
- return nil
- })
- if err != nil {
- span.RecordError(err)
- span.SetStatus(codes.Error, err.Error())
- slog.ErrorContext(ctx, "event not stored", "topic", topic, "error", err)
- return fmt.Errorf("store event: %w", err)
}
return nil
}
diff --git a/pkg/events/bus_test.go b/pkg/events/bus_test.go
index 4f4e7acff..dc781cc03 100644
--- a/pkg/events/bus_test.go
+++ b/pkg/events/bus_test.go
@@ -53,6 +53,16 @@ func onlyDelivery(t *testing.T, bus *Bus, status models.EventDeliveryStatus) mod
return rows[0]
}
+// publish stores the event in its own transaction and fails the test if that fails.
+func publish(t *testing.T, ctx context.Context, bus *Bus, topic EventType, payload any) {
+ t.Helper()
+ if err := db.TxFn(ctx, bus.db, func(tx *sql.Tx) error {
+ return bus.Publish(ctx, tx, topic, payload)
+ }); err != nil {
+ t.Fatalf("publish: %v", err)
+ }
+}
+
func countEvents(t *testing.T, database *sqlx.DB) int {
t.Helper()
var n int
@@ -78,9 +88,7 @@ func TestPublishIsDeliveredToSubscriber(t *testing.T) {
ctx, span := tracing.Start(context.Background(), "test.publish")
wantTrace, _ := tracing.IDsFromContext(ctx)
- if err := bus.Publish(ctx, ComputeClusterUserCreateEvent, clusterUser()); err != nil {
- t.Fatalf("publish: %v", err)
- }
+ publish(t, ctx, bus, ComputeClusterUserCreateEvent, clusterUser())
span.End()
bus.deliverDue(context.Background())
@@ -100,11 +108,31 @@ func TestPublishWritesNothingWithoutSubscription(t *testing.T) {
database := setupTestDB(t)
bus := newBus(t, database)
- if err := bus.Publish(context.Background(), UserCreateEvent, &models.User{ID: "u-1"}); err != nil {
+ publish(t, context.Background(), bus, UserCreateEvent, &models.User{ID: "u-1"})
+ if n := countEvents(t, database); n != 0 {
+ t.Fatalf("expected no stored events, got %d", n)
+ }
+}
+
+// Make sure nothing is stored when the caller's transaction rolls back.
+func TestPublishRollsBackWithCallerTransaction(t *testing.T) {
+ database := setupTestDB(t)
+ bus := newBus(t, database)
+ bus.SubscribeComputeClusterUserCreated(subscriberTest, func(context.Context, models.ComputeClusterUser) error { return nil })
+
+ tx, err := database.BeginTx(context.Background(), nil)
+ if err != nil {
+ t.Fatalf("begin: %v", err)
+ }
+ if err := bus.Publish(context.Background(), tx, ComputeClusterUserCreateEvent, clusterUser()); err != nil {
t.Fatalf("publish: %v", err)
}
+ if err := tx.Rollback(); err != nil {
+ t.Fatalf("rollback: %v", err)
+ }
+
if n := countEvents(t, database); n != 0 {
- t.Fatalf("expected no stored events, got %d", n)
+ t.Fatalf("expected the event to roll back with the transaction, got %d stored", n)
}
}
@@ -124,9 +152,7 @@ func TestFailedDeliveryIsRetriedAfterBackoff(t *testing.T) {
return handler(ctx, cu)
})
- if err := bus.Publish(context.Background(), ComputeClusterUserCreateEvent, clusterUser()); err != nil {
- t.Fatalf("publish: %v", err)
- }
+ publish(t, context.Background(), bus, ComputeClusterUserCreateEvent, clusterUser())
// The second pass runs before the backoff has passed, so it must skip the row.
bus.deliverDue(context.Background())
bus.deliverDue(context.Background())
@@ -154,9 +180,7 @@ func TestDeliveryFailsAtAttemptCap(t *testing.T) {
return errors.New("registry 503")
})
- if err := bus.Publish(context.Background(), ComputeClusterUserCreateEvent, clusterUser()); err != nil {
- t.Fatalf("publish: %v", err)
- }
+ publish(t, context.Background(), bus, ComputeClusterUserCreateEvent, clusterUser())
// Jump the row to its last allowed attempt instead of failing it nine times.
pending := onlyDelivery(t, bus, models.EventDeliveryPending)
if err := db.TxFn(context.Background(), database, func(tx *sql.Tx) error {
@@ -179,9 +203,7 @@ func TestPermanentErrorFailsWithoutRetry(t *testing.T) {
return fmt.Errorf("%w: bad row", ErrPermanent)
})
- if err := bus.Publish(context.Background(), ComputeClusterUserCreateEvent, clusterUser()); err != nil {
- t.Fatalf("publish: %v", err)
- }
+ publish(t, context.Background(), bus, ComputeClusterUserCreateEvent, clusterUser())
bus.deliverDue(context.Background())
if d := onlyDelivery(t, bus, models.EventDeliveryFailed); d.Attempts != 1 {
@@ -199,9 +221,7 @@ func TestEventPublishedBeforeSubscriberLoadsIsDelivered(t *testing.T) {
// A fresh process: the subscription is saved, the handler is not loaded yet.
bus := newBus(t, database)
- if err := bus.Publish(context.Background(), ComputeClusterUserCreateEvent, clusterUser()); err != nil {
- t.Fatalf("publish: %v", err)
- }
+ publish(t, context.Background(), bus, ComputeClusterUserCreateEvent, clusterUser())
bus.deliverDue(context.Background())
if d := onlyDelivery(t, bus, models.EventDeliveryPending); d.Attempts != 0 {
t.Fatalf("expected the row to wait untouched, got %d attempts", d.Attempts)
@@ -229,9 +249,7 @@ func TestUnsubscribeStopsNewDeliveries(t *testing.T) {
if err := bus.Unsubscribe(context.Background(), subscriberTest); err != nil {
t.Fatalf("unsubscribe: %v", err)
}
- if err := bus.Publish(context.Background(), ComputeClusterUserCreateEvent, clusterUser()); err != nil {
- t.Fatalf("publish: %v", err)
- }
+ publish(t, context.Background(), bus, ComputeClusterUserCreateEvent, clusterUser())
if n := countEvents(t, database); n != 0 {
t.Fatalf("expected no stored events after unsubscribe, got %d", n)
}
diff --git a/pkg/service/compute_allocation.go b/pkg/service/compute_allocation.go
index 4c4da06c5..28ff999e0 100644
--- a/pkg/service/compute_allocation.go
+++ b/pkg/service/compute_allocation.go
@@ -87,12 +87,13 @@ func (s *Service) CreateComputeAllocation(ctx context.Context, alloc *models.Com
}
if err := s.inTx(ctx, func(tx *sql.Tx) error {
- return s.allocs.Create(ctx, tx, alloc)
+ if err := s.allocs.Create(ctx, tx, alloc); err != nil {
+ return err
+ }
+ return s.eventBus.Publish(ctx, tx, events.ComputeAllocationCreateEvent, alloc)
}); err != nil {
return nil, fmt.Errorf("create compute allocation: %w", err)
}
-
- s.eventBus.Publish(ctx, events.ComputeAllocationCreateEvent, alloc)
return alloc, nil
}
@@ -133,12 +134,13 @@ func (s *Service) UpdateComputeAllocation(ctx context.Context, alloc *models.Com
return fmt.Errorf("%w: compute allocation id is required", ErrInvalidInput)
}
if err := s.inTx(ctx, func(tx *sql.Tx) error {
- return s.allocs.Update(ctx, tx, alloc)
+ if err := s.allocs.Update(ctx, tx, alloc); err != nil {
+ return err
+ }
+ return s.eventBus.Publish(ctx, tx, events.ComputeAllocationUpdateEvent, alloc)
}); err != nil {
return fmt.Errorf("update compute allocation: %w", err)
}
-
- s.eventBus.Publish(ctx, events.ComputeAllocationUpdateEvent, alloc)
return nil
}
@@ -155,11 +157,12 @@ func (s *Service) DeleteComputeAllocation(ctx context.Context, id string) error
return ErrNotFound
}
if err := s.inTx(ctx, func(tx *sql.Tx) error {
- return s.allocs.Delete(ctx, tx, id)
+ if err := s.allocs.Delete(ctx, tx, id); err != nil {
+ return err
+ }
+ return s.eventBus.Publish(ctx, tx, events.ComputeAllocationDeleteEvent, alloc)
}); err != nil {
return fmt.Errorf("delete compute allocation: %w", err)
}
-
- s.eventBus.Publish(ctx, events.ComputeAllocationDeleteEvent, alloc)
return nil
}
diff --git a/pkg/service/compute_allocation_diff.go b/pkg/service/compute_allocation_diff.go
index ac250b905..881d0cef5 100644
--- a/pkg/service/compute_allocation_diff.go
+++ b/pkg/service/compute_allocation_diff.go
@@ -58,12 +58,13 @@ func (s *Service) CreateComputeAllocationDiff(ctx context.Context, diff *models.
}
if err := s.inTx(ctx, func(tx *sql.Tx) error {
- return s.allocDiffs.Create(ctx, tx, diff)
+ if err := s.allocDiffs.Create(ctx, tx, diff); err != nil {
+ return err
+ }
+ return s.eventBus.Publish(ctx, tx, events.ComputeAllocationDiffCreateEvent, diff)
}); err != nil {
return nil, fmt.Errorf("create compute allocation diff: %w", err)
}
-
- s.eventBus.Publish(ctx, events.ComputeAllocationDiffCreateEvent, diff)
return diff, nil
}
@@ -124,11 +125,12 @@ func (s *Service) DeleteComputeAllocationDiff(ctx context.Context, id string) er
return ErrNotFound
}
if err := s.inTx(ctx, func(tx *sql.Tx) error {
- return s.allocDiffs.Delete(ctx, tx, id)
+ if err := s.allocDiffs.Delete(ctx, tx, id); err != nil {
+ return err
+ }
+ return s.eventBus.Publish(ctx, tx, events.ComputeAllocationDiffDeleteEvent, diff)
}); err != nil {
return fmt.Errorf("delete compute allocation diff: %w", err)
}
-
- s.eventBus.Publish(ctx, events.ComputeAllocationDiffDeleteEvent, diff)
return nil
}
diff --git a/pkg/service/compute_allocation_membership.go b/pkg/service/compute_allocation_membership.go
index 8b82152e4..33cf4d274 100644
--- a/pkg/service/compute_allocation_membership.go
+++ b/pkg/service/compute_allocation_membership.go
@@ -67,12 +67,13 @@ func (s *Service) CreateComputeAllocationMembership(ctx context.Context, m *mode
}
if err := s.inTx(ctx, func(tx *sql.Tx) error {
- return s.memberships.Create(ctx, tx, m)
+ if err := s.memberships.Create(ctx, tx, m); err != nil {
+ return err
+ }
+ return s.eventBus.Publish(ctx, tx, events.ComputeAllocationMembershipCreateEvent, m)
}); err != nil {
return nil, fmt.Errorf("create compute allocation membership: %w", err)
}
-
- s.eventBus.Publish(ctx, events.ComputeAllocationMembershipCreateEvent, m)
return m, nil
}
@@ -171,12 +172,13 @@ func (s *Service) UpdateComputeAllocationMembership(ctx context.Context, m *mode
}
if err := s.inTx(ctx, func(tx *sql.Tx) error {
- return s.memberships.Update(ctx, tx, m)
+ if err := s.memberships.Update(ctx, tx, m); err != nil {
+ return err
+ }
+ return s.eventBus.Publish(ctx, tx, events.ComputeAllocationMembershipUpdateEvent, m)
}); err != nil {
return nil, fmt.Errorf("update compute allocation membership: %w", err)
}
-
- s.eventBus.Publish(ctx, events.ComputeAllocationMembershipUpdateEvent, m)
return m, nil
}
@@ -198,12 +200,13 @@ func (s *Service) UpdateMembershipStatus(ctx context.Context, id string, status
}
existing.MembershipStatus = status
if err := s.inTx(ctx, func(tx *sql.Tx) error {
- return s.memberships.Update(ctx, tx, existing)
+ if err := s.memberships.Update(ctx, tx, existing); err != nil {
+ return err
+ }
+ return s.eventBus.Publish(ctx, tx, events.ComputeAllocationMembershipUpdateEvent, existing)
}); err != nil {
return nil, fmt.Errorf("update compute allocation membership status: %w", err)
}
-
- s.eventBus.Publish(ctx, events.ComputeAllocationMembershipUpdateEvent, existing)
return existing, nil
}
@@ -220,11 +223,12 @@ func (s *Service) DeleteComputeAllocationMembership(ctx context.Context, id stri
return ErrNotFound
}
if err := s.inTx(ctx, func(tx *sql.Tx) error {
- return s.memberships.Delete(ctx, tx, id)
+ if err := s.memberships.Delete(ctx, tx, id); err != nil {
+ return err
+ }
+ return s.eventBus.Publish(ctx, tx, events.ComputeAllocationMembershipDeleteEvent, existing)
}); err != nil {
return fmt.Errorf("delete compute allocation membership: %w", err)
}
-
- s.eventBus.Publish(ctx, events.ComputeAllocationMembershipDeleteEvent, existing)
return nil
}
diff --git a/pkg/service/compute_allocation_membership_resource_override.go b/pkg/service/compute_allocation_membership_resource_override.go
index d4e6cc355..437e8449d 100644
--- a/pkg/service/compute_allocation_membership_resource_override.go
+++ b/pkg/service/compute_allocation_membership_resource_override.go
@@ -72,12 +72,13 @@ func (s *Service) CreateComputeAllocationMembershipResourceOverride(ctx context.
}
if err := s.inTx(ctx, func(tx *sql.Tx) error {
- return s.membershipOverrides.Create(ctx, tx, o)
+ if err := s.membershipOverrides.Create(ctx, tx, o); err != nil {
+ return err
+ }
+ return s.eventBus.Publish(ctx, tx, events.ComputeAllocationMembershipResourceOverrideCreateEvent, o)
}); err != nil {
return nil, fmt.Errorf("create membership resource override: %w", err)
}
-
- s.eventBus.Publish(ctx, events.ComputeAllocationMembershipResourceOverrideCreateEvent, o)
return o, nil
}
@@ -165,12 +166,13 @@ func (s *Service) UpdateComputeAllocationMembershipResourceOverride(ctx context.
}
if err := s.inTx(ctx, func(tx *sql.Tx) error {
- return s.membershipOverrides.Update(ctx, tx, o)
+ if err := s.membershipOverrides.Update(ctx, tx, o); err != nil {
+ return err
+ }
+ return s.eventBus.Publish(ctx, tx, events.ComputeAllocationMembershipResourceOverrideUpdateEvent, o)
}); err != nil {
return nil, fmt.Errorf("update membership resource override: %w", err)
}
-
- s.eventBus.Publish(ctx, events.ComputeAllocationMembershipResourceOverrideUpdateEvent, o)
return o, nil
}
@@ -187,11 +189,12 @@ func (s *Service) DeleteComputeAllocationMembershipResourceOverride(ctx context.
return ErrNotFound
}
if err := s.inTx(ctx, func(tx *sql.Tx) error {
- return s.membershipOverrides.Delete(ctx, tx, id)
+ if err := s.membershipOverrides.Delete(ctx, tx, id); err != nil {
+ return err
+ }
+ return s.eventBus.Publish(ctx, tx, events.ComputeAllocationMembershipResourceOverrideDeleteEvent, existing)
}); err != nil {
return fmt.Errorf("delete membership resource override: %w", err)
}
-
- s.eventBus.Publish(ctx, events.ComputeAllocationMembershipResourceOverrideDeleteEvent, existing)
return nil
}
diff --git a/pkg/service/compute_allocation_resource.go b/pkg/service/compute_allocation_resource.go
index acd8b5177..6afeab039 100644
--- a/pkg/service/compute_allocation_resource.go
+++ b/pkg/service/compute_allocation_resource.go
@@ -52,12 +52,13 @@ func (s *Service) CreateComputeAllocationResource(ctx context.Context, resource
}
if err := s.inTx(ctx, func(tx *sql.Tx) error {
- return s.resources.Create(ctx, tx, resource)
+ if err := s.resources.Create(ctx, tx, resource); err != nil {
+ return err
+ }
+ return s.eventBus.Publish(ctx, tx, events.ComputeAllocationResourceCreateEvent, resource)
}); err != nil {
return nil, fmt.Errorf("create compute allocation resource: %w", err)
}
-
- s.eventBus.Publish(ctx, events.ComputeAllocationResourceCreateEvent, resource)
return resource, nil
}
@@ -135,12 +136,13 @@ func (s *Service) UpdateComputeAllocationResource(ctx context.Context, resource
return fmt.Errorf("%w: compute allocation resource id is required", ErrInvalidInput)
}
if err := s.inTx(ctx, func(tx *sql.Tx) error {
- return s.resources.Update(ctx, tx, resource)
+ if err := s.resources.Update(ctx, tx, resource); err != nil {
+ return err
+ }
+ return s.eventBus.Publish(ctx, tx, events.ComputeAllocationResourceUpdateEvent, resource)
}); err != nil {
return fmt.Errorf("update compute allocation resource: %w", err)
}
-
- s.eventBus.Publish(ctx, events.ComputeAllocationResourceUpdateEvent, resource)
return nil
}
@@ -157,11 +159,12 @@ func (s *Service) DeleteComputeAllocationResource(ctx context.Context, id string
return ErrNotFound
}
if err := s.inTx(ctx, func(tx *sql.Tx) error {
- return s.resources.Delete(ctx, tx, id)
+ if err := s.resources.Delete(ctx, tx, id); err != nil {
+ return err
+ }
+ return s.eventBus.Publish(ctx, tx, events.ComputeAllocationResourceDeleteEvent, resource)
}); err != nil {
return fmt.Errorf("delete compute allocation resource: %w", err)
}
-
- s.eventBus.Publish(ctx, events.ComputeAllocationResourceDeleteEvent, resource)
return nil
}
diff --git a/pkg/service/compute_allocation_resource_mapping.go b/pkg/service/compute_allocation_resource_mapping.go
index e6d154ec0..acef8f82f 100644
--- a/pkg/service/compute_allocation_resource_mapping.go
+++ b/pkg/service/compute_allocation_resource_mapping.go
@@ -71,12 +71,13 @@ func (s *Service) AttachResourceToAllocation(ctx context.Context, allocationID,
ResourceTime: resourceTime,
}
if err := s.inTx(ctx, func(tx *sql.Tx) error {
- return s.resourceMappings.Create(ctx, tx, mapping)
+ if err := s.resourceMappings.Create(ctx, tx, mapping); err != nil {
+ return err
+ }
+ return s.eventBus.Publish(ctx, tx, events.ComputeAllocationResourceMappingCreateEvent, mapping)
}); err != nil {
return nil, fmt.Errorf("attach resource to allocation: %w", err)
}
-
- s.eventBus.Publish(ctx, events.ComputeAllocationResourceMappingCreateEvent, mapping)
return mapping, nil
}
@@ -104,12 +105,13 @@ func (s *Service) UpdateAllocationResourceMapping(ctx context.Context, allocatio
existing.ResourceAmount = resourceAmount
existing.ResourceTime = resourceTime
if err := s.inTx(ctx, func(tx *sql.Tx) error {
- return s.resourceMappings.Update(ctx, tx, existing)
+ if err := s.resourceMappings.Update(ctx, tx, existing); err != nil {
+ return err
+ }
+ return s.eventBus.Publish(ctx, tx, events.ComputeAllocationResourceMappingUpdateEvent, existing)
}); err != nil {
return nil, fmt.Errorf("update allocation resource mapping: %w", err)
}
-
- s.eventBus.Publish(ctx, events.ComputeAllocationResourceMappingUpdateEvent, existing)
return existing, nil
}
@@ -130,12 +132,13 @@ func (s *Service) DetachResourceFromAllocation(ctx context.Context, allocationID
}
if err := s.inTx(ctx, func(tx *sql.Tx) error {
- return s.resourceMappings.DeleteByPair(ctx, tx, allocationID, resourceID)
+ if err := s.resourceMappings.DeleteByPair(ctx, tx, allocationID, resourceID); err != nil {
+ return err
+ }
+ return s.eventBus.Publish(ctx, tx, events.ComputeAllocationResourceMappingDeleteEvent, existing)
}); err != nil {
return fmt.Errorf("detach resource from allocation: %w", err)
}
-
- s.eventBus.Publish(ctx, events.ComputeAllocationResourceMappingDeleteEvent, existing)
return nil
}
diff --git a/pkg/service/compute_cluster.go b/pkg/service/compute_cluster.go
index 382f22efc..c1388bbcd 100644
--- a/pkg/service/compute_cluster.go
+++ b/pkg/service/compute_cluster.go
@@ -46,12 +46,13 @@ func (s *Service) CreateComputeCluster(ctx context.Context, cluster *models.Comp
}
if err := s.inTx(ctx, func(tx *sql.Tx) error {
- return s.clusters.Create(ctx, tx, cluster)
+ if err := s.clusters.Create(ctx, tx, cluster); err != nil {
+ return err
+ }
+ return s.eventBus.Publish(ctx, tx, events.ComputeClusterCreateEvent, cluster)
}); err != nil {
return nil, fmt.Errorf("create compute cluster: %w", err)
}
-
- s.eventBus.Publish(ctx, events.ComputeClusterCreateEvent, cluster)
return cluster, nil
}
@@ -98,12 +99,13 @@ func (s *Service) UpdateComputeCluster(ctx context.Context, cluster *models.Comp
return fmt.Errorf("%w: compute cluster name is required", ErrInvalidInput)
}
if err := s.inTx(ctx, func(tx *sql.Tx) error {
- return s.clusters.Update(ctx, tx, cluster)
+ if err := s.clusters.Update(ctx, tx, cluster); err != nil {
+ return err
+ }
+ return s.eventBus.Publish(ctx, tx, events.ComputeClusterUpdateEvent, cluster)
}); err != nil {
return fmt.Errorf("update compute cluster: %w", err)
}
-
- s.eventBus.Publish(ctx, events.ComputeClusterUpdateEvent, cluster)
return nil
}
@@ -120,11 +122,12 @@ func (s *Service) DeleteComputeCluster(ctx context.Context, id string) error {
return ErrNotFound
}
if err := s.inTx(ctx, func(tx *sql.Tx) error {
- return s.clusters.Delete(ctx, tx, id)
+ if err := s.clusters.Delete(ctx, tx, id); err != nil {
+ return err
+ }
+ return s.eventBus.Publish(ctx, tx, events.ComputeClusterDeleteEvent, cluster)
}); err != nil {
return fmt.Errorf("delete compute cluster: %w", err)
}
-
- s.eventBus.Publish(ctx, events.ComputeClusterDeleteEvent, cluster)
return nil
}
diff --git a/pkg/service/compute_cluster_user.go b/pkg/service/compute_cluster_user.go
index a29a1b8f3..950a79552 100644
--- a/pkg/service/compute_cluster_user.go
+++ b/pkg/service/compute_cluster_user.go
@@ -91,7 +91,10 @@ func (s *Service) CreateComputeClusterUser(ctx context.Context, cu *models.Compu
}
if err := s.inTx(ctx, func(tx *sql.Tx) error {
- return s.clusterUsers.Create(ctx, tx, cu)
+ if err := s.clusterUsers.Create(ctx, tx, cu); err != nil {
+ return err
+ }
+ return s.eventBus.Publish(ctx, tx, events.ComputeClusterUserCreateEvent, cu)
}); err != nil {
switch {
case isLocalUsernameDuplicate(err):
@@ -103,8 +106,6 @@ func (s *Service) CreateComputeClusterUser(ctx context.Context, cu *models.Compu
return nil, fmt.Errorf("create compute cluster user: %w", err)
}
}
-
- s.eventBus.Publish(ctx, events.ComputeClusterUserCreateEvent, cu)
return cu, nil
}
@@ -216,12 +217,13 @@ func (s *Service) UpdateComputeClusterUser(ctx context.Context, cu *models.Compu
cu.LocalUsername = existing.LocalUsername
}
if err := s.inTx(ctx, func(tx *sql.Tx) error {
- return s.clusterUsers.Update(ctx, tx, cu)
+ if err := s.clusterUsers.Update(ctx, tx, cu); err != nil {
+ return err
+ }
+ return s.eventBus.Publish(ctx, tx, events.ComputeClusterUserUpdateEvent, cu)
}); err != nil {
return fmt.Errorf("update compute cluster user: %w", err)
}
-
- s.eventBus.Publish(ctx, events.ComputeClusterUserUpdateEvent, cu)
return nil
}
@@ -259,11 +261,12 @@ func (s *Service) DeleteComputeClusterUser(ctx context.Context, id string) error
return ErrNotFound
}
if err := s.inTx(ctx, func(tx *sql.Tx) error {
- return s.clusterUsers.Delete(ctx, tx, id)
+ if err := s.clusterUsers.Delete(ctx, tx, id); err != nil {
+ return err
+ }
+ return s.eventBus.Publish(ctx, tx, events.ComputeClusterUserDeleteEvent, cu)
}); err != nil {
return fmt.Errorf("delete compute cluster user: %w", err)
}
-
- s.eventBus.Publish(ctx, events.ComputeClusterUserDeleteEvent, cu)
return nil
}
diff --git a/pkg/service/identity_resolver.go b/pkg/service/identity_resolver.go
index bc59a3bcb..5b80799bc 100644
--- a/pkg/service/identity_resolver.go
+++ b/pkg/service/identity_resolver.go
@@ -188,18 +188,20 @@ func (s *Service) linkBySub(ctx context.Context, claims *identity.Claims) (*mode
if err := s.users.UpdateStatus(ctx, tx, user.ID, models.UserActive); err != nil {
return fmt.Errorf("activate user: %w", err)
}
- return s.writeIdentityAuditTx(ctx, tx, user.ID, map[string]any{
+ if err := s.writeIdentityAuditTx(ctx, tx, user.ID, map[string]any{
"oidc_sub": claims.Sub,
"email": claims.Email,
"source": "oidc_email_fallback",
- })
+ }); err != nil {
+ return err
+ }
+ // Publish the event indicating the user account became active, so the `sub` can be added to the registry.
+ return s.eventBus.Publish(ctx, tx, events.UserIdentityCreateEvent, binding)
}); err != nil {
return nil, err
}
slog.Info("identity linked via email fallback", "user_id", user.ID, "email", claims.Email)
- // Publish the event indicating the user account became active, so the `sub` can be added to the registry.
- s.eventBus.Publish(ctx, events.UserIdentityCreateEvent, binding)
user.Status = models.UserActive
return user, nil
}
diff --git a/pkg/service/organization.go b/pkg/service/organization.go
index bf8b446b3..67f6508cb 100644
--- a/pkg/service/organization.go
+++ b/pkg/service/organization.go
@@ -48,12 +48,13 @@ func (s *Service) CreateOrganization(ctx context.Context, org *models.Organizati
}
if err := s.inTx(ctx, func(tx *sql.Tx) error {
- return s.orgs.Create(ctx, tx, org)
+ if err := s.orgs.Create(ctx, tx, org); err != nil {
+ return err
+ }
+ return s.eventBus.Publish(ctx, tx, events.OrganizationCreateEvent, org)
}); err != nil {
return nil, fmt.Errorf("create organization: %w", err)
}
-
- s.eventBus.Publish(ctx, events.OrganizationCreateEvent, org)
return org, nil
}
@@ -97,12 +98,13 @@ func (s *Service) UpdateOrganization(ctx context.Context, org *models.Organizati
return fmt.Errorf("%w: organization id is required", ErrInvalidInput)
}
if err := s.inTx(ctx, func(tx *sql.Tx) error {
- return s.orgs.Update(ctx, tx, org)
+ if err := s.orgs.Update(ctx, tx, org); err != nil {
+ return err
+ }
+ return s.eventBus.Publish(ctx, tx, events.OrganizationUpdateEvent, org)
}); err != nil {
return fmt.Errorf("update organization: %w", err)
}
-
- s.eventBus.Publish(ctx, events.OrganizationUpdateEvent, org)
return nil
}
@@ -119,11 +121,12 @@ func (s *Service) DeleteOrganization(ctx context.Context, id string) error {
return ErrNotFound
}
if err := s.inTx(ctx, func(tx *sql.Tx) error {
- return s.orgs.Delete(ctx, tx, id)
+ if err := s.orgs.Delete(ctx, tx, id); err != nil {
+ return err
+ }
+ return s.eventBus.Publish(ctx, tx, events.OrganizationDeleteEvent, org)
}); err != nil {
return fmt.Errorf("delete organization: %w", err)
}
-
- s.eventBus.Publish(ctx, events.OrganizationDeleteEvent, org)
return nil
}
diff --git a/pkg/service/project.go b/pkg/service/project.go
index ed3117eee..e385ebea2 100644
--- a/pkg/service/project.go
+++ b/pkg/service/project.go
@@ -65,12 +65,13 @@ func (s *Service) CreateProject(ctx context.Context, project *models.Project) (*
}
if err := s.inTx(ctx, func(tx *sql.Tx) error {
- return s.projs.Create(ctx, tx, project)
+ if err := s.projs.Create(ctx, tx, project); err != nil {
+ return err
+ }
+ return s.eventBus.Publish(ctx, tx, events.ProjectCreateEvent, project)
}); err != nil {
return nil, fmt.Errorf("create project: %w", err)
}
-
- s.eventBus.Publish(ctx, events.ProjectCreateEvent, project)
return project, nil
}
@@ -199,12 +200,13 @@ func (s *Service) UpdateProject(ctx context.Context, project *models.Project) er
project.CreatedTime = existing.CreatedTime
}
if err := s.inTx(ctx, func(tx *sql.Tx) error {
- return s.projs.Update(ctx, tx, project)
+ if err := s.projs.Update(ctx, tx, project); err != nil {
+ return err
+ }
+ return s.eventBus.Publish(ctx, tx, events.ProjectUpdateEvent, project)
}); err != nil {
return fmt.Errorf("update project: %w", err)
}
-
- s.eventBus.Publish(ctx, events.ProjectUpdateEvent, project)
return nil
}
@@ -224,14 +226,15 @@ func (s *Service) UpdateProjectStatus(ctx context.Context, id string, status mod
if existing == nil {
return nil, ErrNotFound
}
+ existing.Status = status
if err := s.inTx(ctx, func(tx *sql.Tx) error {
- return s.projs.UpdateStatus(ctx, tx, id, status)
+ if err := s.projs.UpdateStatus(ctx, tx, id, status); err != nil {
+ return err
+ }
+ return s.eventBus.Publish(ctx, tx, events.ProjectUpdateEvent, existing)
}); err != nil {
return nil, fmt.Errorf("update project status: %w", err)
}
- existing.Status = status
-
- s.eventBus.Publish(ctx, events.ProjectUpdateEvent, existing)
return existing, nil
}
@@ -248,11 +251,12 @@ func (s *Service) DeleteProject(ctx context.Context, id string) error {
return ErrNotFound
}
if err := s.inTx(ctx, func(tx *sql.Tx) error {
- return s.projs.Delete(ctx, tx, id)
+ if err := s.projs.Delete(ctx, tx, id); err != nil {
+ return err
+ }
+ return s.eventBus.Publish(ctx, tx, events.ProjectDeleteEvent, project)
}); err != nil {
return fmt.Errorf("delete project: %w", err)
}
-
- s.eventBus.Publish(ctx, events.ProjectDeleteEvent, project)
return nil
}
diff --git a/pkg/service/user.go b/pkg/service/user.go
index f17216ecc..2a7e9e117 100644
--- a/pkg/service/user.go
+++ b/pkg/service/user.go
@@ -33,12 +33,13 @@ func (s *Service) CreateUser(ctx context.Context, user *models.User) (*models.Us
return nil, err
}
if err := s.inTx(ctx, func(tx *sql.Tx) error {
- return s.users.Create(ctx, tx, user)
+ if err := s.users.Create(ctx, tx, user); err != nil {
+ return err
+ }
+ return s.eventBus.Publish(ctx, tx, events.UserCreateEvent, user)
}); err != nil {
return nil, fmt.Errorf("create user: %w", err)
}
-
- s.eventBus.Publish(ctx, events.UserCreateEvent, user)
return user, nil
}
@@ -185,12 +186,13 @@ func (s *Service) UpdateUser(ctx context.Context, user *models.User) error {
}
if err := s.inTx(ctx, func(tx *sql.Tx) error {
- return s.users.Update(ctx, tx, user)
+ if err := s.users.Update(ctx, tx, user); err != nil {
+ return err
+ }
+ return s.eventBus.Publish(ctx, tx, events.UserUpdateEvent, user)
}); err != nil {
return fmt.Errorf("update user: %w", err)
}
-
- s.eventBus.Publish(ctx, events.UserUpdateEvent, user)
return nil
}
@@ -210,14 +212,15 @@ func (s *Service) UpdateUserStatus(ctx context.Context, id string, status models
if existing == nil {
return nil, ErrNotFound
}
+ existing.Status = status
if err := s.inTx(ctx, func(tx *sql.Tx) error {
- return s.users.UpdateStatus(ctx, tx, id, status)
+ if err := s.users.UpdateStatus(ctx, tx, id, status); err != nil {
+ return err
+ }
+ return s.eventBus.Publish(ctx, tx, events.UserUpdateEvent, existing)
}); err != nil {
return nil, fmt.Errorf("update user status: %w", err)
}
- existing.Status = status
-
- s.eventBus.Publish(ctx, events.UserUpdateEvent, existing)
return existing, nil
}
@@ -234,11 +237,12 @@ func (s *Service) DeleteUser(ctx context.Context, id string) error {
return ErrNotFound
}
if err := s.inTx(ctx, func(tx *sql.Tx) error {
- return s.users.Delete(ctx, tx, id)
+ if err := s.users.Delete(ctx, tx, id); err != nil {
+ return err
+ }
+ return s.eventBus.Publish(ctx, tx, events.UserDeleteEvent, existing)
}); err != nil {
return fmt.Errorf("delete user: %w", err)
}
-
- s.eventBus.Publish(ctx, events.UserDeleteEvent, existing)
return nil
}
diff --git a/pkg/service/user_identity.go b/pkg/service/user_identity.go
index 2fc218a55..dbffc363f 100644
--- a/pkg/service/user_identity.go
+++ b/pkg/service/user_identity.go
@@ -59,12 +59,13 @@ func (s *Service) CreateUserIdentity(ctx context.Context, e *models.UserIdentity
}
if err := s.inTx(ctx, func(tx *sql.Tx) error {
- return s.userIdentities.Create(ctx, tx, e)
+ if err := s.userIdentities.Create(ctx, tx, e); err != nil {
+ return err
+ }
+ return s.eventBus.Publish(ctx, tx, events.UserIdentityCreateEvent, e)
}); err != nil {
return nil, fmt.Errorf("create user identity: %w", err)
}
-
- s.eventBus.Publish(ctx, events.UserIdentityCreateEvent, e)
return e, nil
}
@@ -158,12 +159,13 @@ func (s *Service) UpdateUserIdentity(ctx context.Context, e *models.UserIdentity
e.Metadata = existing.Metadata
}
if err := s.inTx(ctx, func(tx *sql.Tx) error {
- return s.userIdentities.Update(ctx, tx, e)
+ if err := s.userIdentities.Update(ctx, tx, e); err != nil {
+ return err
+ }
+ return s.eventBus.Publish(ctx, tx, events.UserIdentityUpdateEvent, e)
}); err != nil {
return fmt.Errorf("update user identity: %w", err)
}
-
- s.eventBus.Publish(ctx, events.UserIdentityUpdateEvent, e)
return nil
}
@@ -180,11 +182,12 @@ func (s *Service) DeleteUserIdentity(ctx context.Context, id string) error {
return ErrNotFound
}
if err := s.inTx(ctx, func(tx *sql.Tx) error {
- return s.userIdentities.Delete(ctx, tx, id)
+ if err := s.userIdentities.Delete(ctx, tx, id); err != nil {
+ return err
+ }
+ return s.eventBus.Publish(ctx, tx, events.UserIdentityDeleteEvent, e)
}); err != nil {
return fmt.Errorf("delete user identity: %w", err)
}
-
- s.eventBus.Publish(ctx, events.UserIdentityDeleteEvent, e)
return nil
}
diff --git a/pkg/service/user_merge.go b/pkg/service/user_merge.go
index e2c3d5946..ab9976cf6 100644
--- a/pkg/service/user_merge.go
+++ b/pkg/service/user_merge.go
@@ -94,14 +94,17 @@ func (s *Service) MergeUsers(ctx context.Context, survivingID, retiringID string
if err := s.projMemberships.ReassignUser(ctx, tx, retiringID, survivingID); err != nil {
return fmt.Errorf("reassign project memberships: %w", err)
}
- return s.users.UpdateStatus(ctx, tx, retiringID, models.UserMerged)
+ if err := s.users.UpdateStatus(ctx, tx, retiringID, models.UserMerged); err != nil {
+ return err
+ }
+ retiring.Status = models.UserMerged
+ if err := s.eventBus.Publish(ctx, tx, events.UserUpdateEvent, retiring); err != nil {
+ return err
+ }
+ return s.eventBus.Publish(ctx, tx, events.UserUpdateEvent, survivor)
}); err != nil {
return nil, fmt.Errorf("merge users: %w", err)
}
-
- retiring.Status = models.UserMerged
- s.eventBus.Publish(ctx, events.UserUpdateEvent, retiring)
- s.eventBus.Publish(ctx, events.UserUpdateEvent, survivor)
return survivor, nil
}
diff --git a/pkg/service/user_onboard.go b/pkg/service/user_onboard.go
index 4c3755500..516342482 100644
--- a/pkg/service/user_onboard.go
+++ b/pkg/service/user_onboard.go
@@ -252,16 +252,18 @@ func (s *Service) OnboardUser(ctx context.Context, in OnboardUserInput) (*models
return fmt.Errorf("create allocation membership: %w", err)
}
}
+
+ if err := s.eventBus.Publish(ctx, tx, events.UserCreateEvent, created); err != nil {
+ return err
+ }
+ if needsCluster {
+ return s.eventBus.Publish(ctx, tx, events.ComputeClusterUserCreateEvent, clusterUser)
+ }
return nil
}); err != nil {
return nil, err
}
- // Published after commit so a subscriber never provisions a rolled-back row.
- s.eventBus.Publish(ctx, events.UserCreateEvent, created)
- if needsCluster {
- s.eventBus.Publish(ctx, events.ComputeClusterUserCreateEvent, clusterUser)
- }
if in.ClusterAdmin {
slog.Info("cluster admin account created", "user_id", created.ID, "cluster_id", clusterUser.ComputeClusterID, "actor_id", in.OnboardedBy)
}