fix: keep existing privileges when User.Added is processed late

Setup() registers one transient consumer per routing key, and each mints its
own queue drained by its own goroutine, so User.Added and Privilege.Added for the
same company can be processed in either order. Process(*UserAdded) replaced the
company entry with empty privileges, so a Privilege.Added handled first lost its
privilege until the next Fetch(), which only runs at service start.

Create the entry only when it is missing instead, matching authz-service's own
aggregate and read view, which both leave an existing user entry alone. This also
makes a User.Added redelivery harmless.

Seen as authz-service #826 acceptance-test failures: a new company's Admin grant
vanished, so company-service's CreateCompany callback waited out its 30s timeout
and the company page stayed empty.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01XMvdB7bcwn1CrQKM4dCshM

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"))