fix: keep existing privileges when User.Added is processed late (#331)

CLAUDE.md

@@ -43,4 +43,4 @@ The `CompanyPrivileges` struct contains permission flags:
 
 ### Event Handling
 
-Registers per-replica (transient) go-messaging-amqp consumers for privilege update events from the authz-service (`Setup()`), keeping the local privilege cache up-to-date. Don't combine `Setup()` with go-messaging-amqp's `WithReconnect`: a reconnect declares new per-replica queues, so revocations published during the outage are lost unless `Fetch()` runs again. Services exit on connection loss (`CloseListener`) and re-fetch on start.
+Registers per-replica (transient) go-messaging-amqp consumers for privilege update events from the authz-service (`Setup()`), keeping the local privilege cache up-to-date. Each routing key gets its own queue, so events for one company arrive in any order — `Process` must never discard state it didn't itself record. `User.Added` therefore creates the company entry only when it is missing: a `Privilege.Added` processed ahead of it would otherwise be overwritten, and nothing re-reads the privilege until the next `Fetch()`. The converse doesn't hold yet — `setPrivileges` still creates an entry for a company it never saw a `User.Added` for, so a `Privilege.*` delivered after `User.Removed` resurrects an all-false membership. Don't combine `Setup()` with go-messaging-amqp's `WithReconnect`: a reconnect declares new per-replica queues, so revocations published during the outage are lost unless `Fetch()` runs again. Services exit on connection loss (`CloseListener`) and re-fetch on start.

client.go

@@ -123,12 +123,17 @@ func (h *PrivilegeHandler) Process(msg any) error {
 
 	switch ev := msg.(type) {
 	case *UserAdded:
-		if priv, exists := h.privileges[ev.Email]; exists {
+		// Keep the privileges already recorded for the company. Each routing key has
+		// its own transient queue, so a Privilege.Added published after this event can
+		// be processed before it; overwriting the entry here would drop that privilege
+		// until the next Fetch, which only runs at start.
+		priv, exists := h.privileges[ev.Email]
+		if !exists {
+			priv = map[string]*CompanyPrivileges{}
+			h.privileges[ev.Email] = priv
+		}
+		if _, exists := priv[ev.CompanyID]; !exists {
 			priv[ev.CompanyID] = &CompanyPrivileges{}
-		} else {
-			h.privileges[ev.Email] = map[string]*CompanyPrivileges{
-				ev.CompanyID: {},
-			}
 		}
 		return nil
 	case *UserRemoved:

client_test.go

@@ -12,6 +12,7 @@ import (
 	goamqp "codeberg.org/messaging/go-messaging-amqp"
 	spec "codeberg.org/messaging/messaging"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func TestPrivilegeHandler_Process_InvalidType(t *testing.T) {
@@ -91,6 +92,50 @@ func TestPrivilegeHandler_Process_UserAdded_And_UserRemoved(t *testing.T) {
 	assert.Empty(t, companies)
 }
 
+func TestPrivilegeHandler_Process_UserAdded_Keeps_Existing_Privileges(t *testing.T) {
+	// Each routing key has its own transient queue, so Privilege.Added can be
+	// processed before the User.Added published before it. The privilege must
+	// survive either order: nothing re-reads it until the next Fetch at start.
+	userAdded := &UserAdded{Email: "jim@example.org", CompanyID: "abc-123"}
+	privilegeAdded := &PrivilegeAdded{Email: "jim@example.org", CompanyID: "abc-123", Privilege: PrivilegeAdmin}
+	for name, order := range map[string][]any{
+		"user added first":      {userAdded, privilegeAdded},
+		"privilege added first": {privilegeAdded, userAdded},
+	} {
+		t.Run(name, func(t *testing.T) {
+			handler := New(WithBaseURL("base"))
+			for _, event := range order {
+				require.NoError(t, handler.Process(event))
+			}
+
+			assert.True(t, handler.IsAllowed("jim@example.org", "abc-123", func(privileges CompanyPrivileges) bool {
+				return privileges.Admin
+			}))
+		})
+	}
+}
+
+func TestPrivilegeHandler_Process_UserAdded_After_UserRemoved_Starts_Empty(t *testing.T) {
+	handler := New(WithBaseURL("base"))
+
+	assert.NoError(t, handler.Process(&UserAdded{Email: "jim@example.org", CompanyID: "abc-123"}))
+	assert.NoError(t, handler.Process(&PrivilegeAdded{
+		Email:     "jim@example.org",
+		CompanyID: "abc-123",
+		Privilege: PrivilegeAdmin,
+	}))
+	assert.NoError(t, handler.Process(&UserRemoved{Email: "jim@example.org", CompanyID: "abc-123"}))
+	assert.NoError(t, handler.Process(&UserAdded{Email: "jim@example.org", CompanyID: "abc-123"}))
+
+	// Membership is back, without the privileges the removal took away.
+	assert.Equal(t, []string{"abc-123"}, handler.CompaniesByUser("jim@example.org", func(CompanyPrivileges) bool {
+		return true
+	}))
+	assert.False(t, handler.IsAllowed("jim@example.org", "abc-123", func(privileges CompanyPrivileges) bool {
+		return privileges.Admin
+	}))
+}
+
 func TestPrivilegeHandler_GetCompanies_Email_Not_Found(t *testing.T) {
 	handler := New(WithBaseURL("base"))
 

go.mod

@@ -3,7 +3,7 @@ module gitea.unbound.se/shiny/authz_client
 go 1.26.2
 
 require (
-	codeberg.org/messaging/go-messaging-amqp v0.0.4
+	codeberg.org/messaging/go-messaging-amqp v0.0.5
 	codeberg.org/messaging/messaging v0.0.5
 	github.com/stretchr/testify v1.12.1
 )
@@ -19,7 +19,7 @@ require (
 	github.com/prometheus/client_model v0.6.2 // indirect
 	github.com/prometheus/common v0.66.1 // indirect
 	github.com/prometheus/procfs v0.16.1 // indirect
-	github.com/rabbitmq/amqp091-go v1.12.0 // indirect
+	github.com/rabbitmq/amqp091-go v1.15.0 // indirect
 	go.opentelemetry.io/auto/sdk v1.2.1 // indirect
 	go.opentelemetry.io/otel v1.44.0 // indirect
 	go.opentelemetry.io/otel/metric v1.44.0 // indirect

go.sum

@@ -1,5 +1,5 @@
-codeberg.org/messaging/go-messaging-amqp v0.0.4 h1:MwY/kU1lBdCL7ywKAg2k6aHuJf8f78MRZQSuyrNFvQ8=
+codeberg.org/messaging/go-messaging-amqp v0.0.5 h1:zzcJ+FVWgBPKyTkq9RXF11yvnhs8IN5BALBC+mps1zA=
-codeberg.org/messaging/go-messaging-amqp v0.0.4/go.mod h1:6bSCIkKH0V/oI7vaIq8ttINKYwov9c8ckZrwjfvp7kc=
+codeberg.org/messaging/go-messaging-amqp v0.0.5/go.mod h1:6bSCIkKH0V/oI7vaIq8ttINKYwov9c8ckZrwjfvp7kc=
 codeberg.org/messaging/messaging v0.0.5 h1:/ueH90F4RNPUeeJIwdG9oVhDW7+XjCzwOYq8xc0kfQk=
 codeberg.org/messaging/messaging v0.0.5/go.mod h1:xyWLUcfaVzcN6GWk9uaQsxPVUC2Vm2S89mYZGdiWTWg=
 codeberg.org/messaging/messaging/tck v0.0.3 h1:a4Nr7ytFqEJloTOyVeAtMNgO9Fig1ZUirjW4FVlK0lc=
@@ -39,8 +39,8 @@ github.com/prometheus/common v0.66.1 h1:h5E0h5/Y8niHc5DlaLlWLArTQI7tMrsfQjHV+d9Z
 github.com/prometheus/common v0.66.1/go.mod h1:gcaUsgf3KfRSwHY4dIMXLPV0K/Wg1oZ8+SbZk/HH/dA=
 github.com/prometheus/procfs v0.16.1 h1:hZ15bTNuirocR6u0JZ6BAHHmwS1p8B4P6MRqxtzMyRg=
 github.com/prometheus/procfs v0.16.1/go.mod h1:teAbpZRB1iIAJYREa1LsoWUXykVXA1KlTmWl8x/U+Is=
-github.com/rabbitmq/amqp091-go v1.12.0 h1:V0v14Iqfs+MwHWihJt/nGS5Ulu0vw572b2Co3mwunkI=
+github.com/rabbitmq/amqp091-go v1.15.0 h1:LEQL4/yp48/Wigt6A6XOu18RQRo8ZHtB5I/KZJn+gkw=
-github.com/rabbitmq/amqp091-go v1.12.0/go.mod h1:Hy4jKW5kQART1u+JkDTF9YYOQUHXqMuhrgxOEeS7G4o=
+github.com/rabbitmq/amqp091-go v1.15.0/go.mod h1:Hy4jKW5kQART1u+JkDTF9YYOQUHXqMuhrgxOEeS7G4o=
 github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ=
 github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc=
 github.com/stretchr/testify v1.12.1 h1:EuwCh5fleGS7H32xRwO3wRGT7DxrDhLAT6FF8MpWDWE=