authz_client/client_test.go

package client

import (
	"context"
	"fmt"
	"math"
	"net/http"
	"net/http/httptest"
	"sort"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	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) {
	handler := New(WithBaseURL("base"))

	err := handler.Process("abc")

	assert.EqualError(t, err, "unexpected event type: 'string'")
}

func TestPrivilegeHandler_Process_PrivilegeRemoved(t *testing.T) {
	handler := New(WithBaseURL("base"))

	err := handler.Process(&PrivilegeAdded{
		SequenceNo: 1,
		Email:      "jim@example.org",
		CompanyID:  "abc-123",
		Privilege:  PrivilegeAdmin,
	})
	assert.NoError(t, err)

	companies := handler.CompaniesByUser("jim@example.org", func(privileges CompanyPrivileges) bool {
		return privileges.Admin
	})

	assert.Equal(t, []string{"abc-123"}, companies)

	err = handler.Process(&PrivilegeRemoved{
		SequenceNo: 2,
		Email:      "jim@example.org",
		CompanyID:  "abc-123",
		Privilege:  PrivilegeAdmin,
	})
	assert.NoError(t, err)

	companies = handler.CompaniesByUser("jim@example.org", func(privileges CompanyPrivileges) bool {
		return privileges.Admin
	})

	assert.Empty(t, companies)
}

func TestPrivilegeHandler_Process_UserAdded_And_UserRemoved(t *testing.T) {
	handler := New(WithBaseURL("base"))

	err := handler.Process(&UserAdded{
		SequenceNo: 1,
		Email:      "jim@example.org",
		CompanyID:  "abc-123",
	})
	assert.NoError(t, err)

	err = handler.Process(&UserAdded{
		SequenceNo: 2,
		Email:      "jim@example.org",
		CompanyID:  "abc-456",
	})
	assert.NoError(t, err)

	companies := handler.CompaniesByUser("jim@example.org", func(privileges CompanyPrivileges) bool {
		return true
	})
	sort.Strings(companies)
	assert.Equal(t, []string{"abc-123", "abc-456"}, companies)

	err = handler.Process(&UserRemoved{
		SequenceNo: 3,
		Email:      "jim@example.org",
		CompanyID:  "abc-123",
	})
	assert.NoError(t, err)

	err = handler.Process(&UserRemoved{
		SequenceNo: 4,
		Email:      "jim@example.org",
		CompanyID:  "abc-456",
	})
	assert.NoError(t, err)

	companies = handler.CompaniesByUser("jim@example.org", func(privileges CompanyPrivileges) bool {
		return true
	})
	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{SequenceNo: 1, Email: "jim@example.org", CompanyID: "abc-123"}
	privilegeAdded := &PrivilegeAdded{SequenceNo: 2, 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{SequenceNo: 1, Email: "jim@example.org", CompanyID: "abc-123"}))
	assert.NoError(t, handler.Process(&PrivilegeAdded{
		SequenceNo: 2,
		Email:      "jim@example.org",
		CompanyID:  "abc-123",
		Privilege:  PrivilegeAdmin,
	}))
	assert.NoError(t, handler.Process(&UserRemoved{SequenceNo: 3, Email: "jim@example.org", CompanyID: "abc-123"}))
	assert.NoError(t, handler.Process(&UserAdded{SequenceNo: 4, 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"))

	companies := handler.CompaniesByUser("jim@example.org", func(privileges CompanyPrivileges) bool {
		return true
	})

	assert.Empty(t, companies)
}

func TestPrivilegeHandler_GetCompanies_No_Companies_Found(t *testing.T) {
	handler := New(WithBaseURL("base"))

	err := handler.Process(&UserAdded{
		SequenceNo: 1,
		Email:      "jim@example.org",
		CompanyID:  "abc-123",
	})
	assert.NoError(t, err)

	companies := handler.CompaniesByUser("jim@example.org", func(privileges CompanyPrivileges) bool {
		return privileges.Admin
	})

	assert.Empty(t, companies)

	companies = handler.CompaniesByUser("jim@example.org", func(privileges CompanyPrivileges) bool {
		return true
	})

	assert.Equal(t, []string{"abc-123"}, companies)

	err = handler.Process(&UserRemoved{
		SequenceNo: 2,
		Email:      "jim@example.org",
		CompanyID:  "abc-123",
	})
	assert.NoError(t, err)

	companies = handler.CompaniesByUser("jim@example.org", func(privileges CompanyPrivileges) bool {
		return true
	})
	assert.Empty(t, companies)
}

func TestPrivilegeHandler_GetCompanies_Company_With_Company_Access_Found(t *testing.T) {
	handler := New(WithBaseURL("base"))

	err := handler.Process(&PrivilegeAdded{
		SequenceNo: 1,
		Email:      "jim@example.org",
		CompanyID:  "abc-123",
		Privilege:  PrivilegeCompany,
	})
	assert.NoError(t, err)

	companies := handler.CompaniesByUser("jim@example.org", func(privileges CompanyPrivileges) bool {
		return privileges.Company
	})

	assert.Equal(t, []string{"abc-123"}, companies)
}

func TestPrivilegeHandler_GetCompanies_Company_With_Admin_Access_Found(t *testing.T) {
	handler := New(WithBaseURL("base"))

	err := handler.Process(&PrivilegeAdded{
		SequenceNo: 1,
		Email:      "jim@example.org",
		CompanyID:  "abc-123",
		Privilege:  PrivilegeConsumer,
	})
	assert.NoError(t, err)

	companies := handler.CompaniesByUser("jim@example.org", func(privileges CompanyPrivileges) bool {
		return privileges.Consumer
	})

	assert.Equal(t, []string{"abc-123"}, companies)
}

func TestPrivilegeHandler_IsAllowed_Return_False_If_No_Privileges(t *testing.T) {
	handler := New(WithBaseURL("base"))

	result := handler.IsAllowed("jim@example.org", "abc-123", func(privileges CompanyPrivileges) bool {
		return privileges.Company
	})

	assert.False(t, result)
}

func TestPrivilegeHandler_IsAllowed_Return_True_If_Privilege_Exists(t *testing.T) {
	handler := New(WithBaseURL("base"))

	_ = handler.Process(&PrivilegeAdded{
		SequenceNo: 1,
		Email:      "jim@example.org",
		CompanyID:  "abc-123",
		Privilege:  PrivilegeTime,
	})

	result := handler.IsAllowed("jim@example.org", "abc-123", func(privileges CompanyPrivileges) bool {
		return privileges.Time
	})

	assert.True(t, result)

	_ = handler.Process(&PrivilegeAdded{
		SequenceNo: 2,
		Email:      "jim@example.org",
		CompanyID:  "abc-123",
		Privilege:  PrivilegeInvoicing,
	})

	result = handler.IsAllowed("jim@example.org", "abc-123", func(privileges CompanyPrivileges) bool {
		return privileges.Invoicing
	})

	assert.True(t, result)

	_ = handler.Process(&PrivilegeAdded{
		SequenceNo: 3,
		Email:      "jim@example.org",
		CompanyID:  "abc-123",
		Privilege:  PrivilegeAccounting,
	})

	result = handler.IsAllowed("jim@example.org", "abc-123", func(privileges CompanyPrivileges) bool {
		return privileges.Accounting
	})

	assert.True(t, result)

	_ = handler.Process(&PrivilegeAdded{
		SequenceNo: 4,
		Email:      "jim@example.org",
		CompanyID:  "abc-123",
		Privilege:  PrivilegeSupplier,
	})

	result = handler.IsAllowed("jim@example.org", "abc-123", func(privileges CompanyPrivileges) bool {
		return privileges.Supplier
	})

	assert.True(t, result)

	_ = handler.Process(&PrivilegeAdded{
		SequenceNo: 5,
		Email:      "jim@example.org",
		CompanyID:  "abc-123",
		Privilege:  PrivilegeSalary,
	})

	result = handler.IsAllowed("jim@example.org", "abc-123", func(privileges CompanyPrivileges) bool {
		return privileges.Salary
	})

	assert.True(t, result)
}

func TestPrivilegeHandler_Fetch_Sends_Authorization_Header_When_APIKey_Set(t *testing.T) {
	var receivedAuth string
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		receivedAuth = r.Header.Get("Authorization")
		_, _ = w.Write([]byte("{}"))
	}))
	defer server.Close()

	handler := New(
		WithBaseURL(server.URL),
		WithAPIKey("my-secret-key"),
	)

	err := handler.Fetch()
	assert.NoError(t, err)
	assert.Equal(t, "Bearer my-secret-key", receivedAuth)
}

func TestPrivilegeHandler_Fetch_No_Authorization_Header_Without_APIKey(t *testing.T) {
	var receivedAuth string
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		receivedAuth = r.Header.Get("Authorization")
		_, _ = w.Write([]byte("{}"))
	}))
	defer server.Close()

	handler := New(WithBaseURL(server.URL))

	err := handler.Fetch()
	assert.NoError(t, err)
	assert.Empty(t, receivedAuth)
}

func TestPrivilegeHandler_Fetch_Error_Response(t *testing.T) {
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.WriteHeader(500)
	}))

	baseURL := server.Listener.Addr().String()
	handler := New(WithBaseURL(fmt.Sprintf("http://%s", baseURL)))

	server.Close()

	err := handler.Fetch()
	assert.EqualError(t, err, fmt.Sprintf("Get \"http://%s/authz\": dial tcp %s: connect: connection refused", baseURL, baseURL))
}

func TestPrivilegeHandler_Fetch_Error_Unreadable_Body(t *testing.T) {
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Length", "1")
	}))
	defer server.Close()

	baseURL := server.Listener.Addr().String()
	handler := New(WithBaseURL(fmt.Sprintf("http://%s", baseURL)))

	err := handler.Fetch()
	assert.EqualError(t, err, "unexpected EOF")
}

func TestPrivilegeHandler_Fetch_Error_Broken_JSON(t *testing.T) {
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		_, _ = w.Write([]byte("{abc"))
	}))
	defer server.Close()

	baseURL := server.Listener.Addr().String()
	handler := New(WithBaseURL(fmt.Sprintf("http://%s", baseURL)))

	err := handler.Fetch()
	assert.EqualError(t, err, "invalid character 'a' looking for beginning of object key string")
}

func TestPrivilegeHandler_Fetch_Valid(t *testing.T) {
	privileges := `
{
  "jim@example.org": {
    "00010203-0405-4607-8809-0a0b0c0d0e0f": {
      "admin": false,
      "company": true,
      "consumer": false,
      "time": true,
      "invoicing": true,
      "accounting": false,
      "supplier": false,
      "salary": true
    }
  }
}`
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		_, _ = w.Write([]byte(privileges))
	}))
	defer server.Close()

	baseURL := server.Listener.Addr().String()
	handler := New(WithBaseURL(fmt.Sprintf("http://%s", baseURL)))

	err := handler.Fetch()
	assert.NoError(t, err)
	expectedPrivileges := map[string]map[string]CompanyPrivileges{
		"jim@example.org": {
			"00010203-0405-4607-8809-0a0b0c0d0e0f": {
				Admin:      false,
				Company:    true,
				Consumer:   false,
				Time:       true,
				Invoicing:  true,
				Accounting: false,
				Supplier:   false,
				Salary:     true,
			},
		},
	}
	assert.Equal(t, expectedPrivileges, privilegesOf(handler))
}

func TestPrivilegeHandler_Fetch_Concurrent_Fetches(t *testing.T) {
	privileges := `
{
  "jim@example.org": {
    "00010203-0405-4607-8809-0a0b0c0d0e0f": {
      "admin": false,
      "company": true,
      "consumer": false,
      "time": true,
      "invoicing": true,
      "accounting": false,
      "supplier": false,
      "salary": true
    }
  }
}`
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		_, _ = w.Write([]byte(privileges))
	}))
	defer server.Close()

	baseURL := server.Listener.Addr().String()
	handler := New(WithBaseURL(fmt.Sprintf("http://%s", baseURL)))

	// Run multiple Fetch calls concurrently to test thread-safety
	var wg sync.WaitGroup
	errors := make(chan error, 10)

	for i := 0; i < 10; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			if err := handler.Fetch(); err != nil {
				errors <- err
			}
		}()
	}

	wg.Wait()
	close(errors)

	// Check no errors occurred
	for err := range errors {
		assert.NoError(t, err)
	}

	// Verify privileges were set correctly
	expectedPrivileges := map[string]map[string]CompanyPrivileges{
		"jim@example.org": {
			"00010203-0405-4607-8809-0a0b0c0d0e0f": {
				Admin:      false,
				Company:    true,
				Consumer:   false,
				Time:       true,
				Invoicing:  true,
				Accounting: false,
				Supplier:   false,
				Salary:     true,
			},
		},
	}
	assert.Equal(t, expectedPrivileges, privilegesOf(handler))
}

func TestPrivilegeHandler_Concurrent_Fetch_And_Read(t *testing.T) {
	privileges := `
{
  "jim@example.org": {
    "abc-123": {
      "admin": true,
      "company": true,
      "consumer": false,
      "time": false,
      "invoicing": false,
      "accounting": false,
      "supplier": false,
      "salary": false
    }
  }
}`
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		_, _ = w.Write([]byte(privileges))
	}))
	defer server.Close()

	baseURL := server.Listener.Addr().String()
	handler := New(WithBaseURL(fmt.Sprintf("http://%s", baseURL)))

	var wg sync.WaitGroup
	errors := make(chan error, 100)

	// Start multiple Fetch operations
	for i := 0; i < 10; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			if err := handler.Fetch(); err != nil {
				errors <- err
			}
		}()
	}

	// Concurrently read privileges while Fetch is running
	for i := 0; i < 50; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			_ = handler.CompaniesByUser("jim@example.org", func(privileges CompanyPrivileges) bool {
				return privileges.Admin
			})
		}()
	}

	// Concurrently check privileges while Fetch is running
	for i := 0; i < 50; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			_ = handler.IsAllowed("jim@example.org", "abc-123", func(privileges CompanyPrivileges) bool {
				return privileges.Admin
			})
		}()
	}

	wg.Wait()
	close(errors)

	// Check no errors occurred
	for err := range errors {
		assert.NoError(t, err)
	}

	// Verify privileges are correct after all concurrent operations
	companies := handler.CompaniesByUser("jim@example.org", func(privileges CompanyPrivileges) bool {
		return privileges.Admin
	})
	assert.Equal(t, []string{"abc-123"}, companies)

	isAllowed := handler.IsAllowed("jim@example.org", "abc-123", func(privileges CompanyPrivileges) bool {
		return privileges.Admin && privileges.Company
	})
	assert.True(t, isAllowed)
}

func TestPrivilegeHandler_Concurrent_Process_And_Read(t *testing.T) {
	handler := New(WithBaseURL("base"))

	var wg sync.WaitGroup

	// Concurrently add privileges via Process
	for i := 0; i < 100; i++ {
		wg.Add(1)
		companyID := fmt.Sprintf("company-%d", i%10)
		go func(id string) {
			defer wg.Done()
			_ = handler.Process(&PrivilegeAdded{
				SequenceNo: 1,
				Email:      "jim@example.org",
				CompanyID:  id,
				Privilege:  PrivilegeAdmin,
			})
		}(companyID)
	}

	// Concurrently read privileges while Process is running
	for i := 0; i < 100; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			_ = handler.CompaniesByUser("jim@example.org", func(privileges CompanyPrivileges) bool {
				return privileges.Admin
			})
		}()
	}

	wg.Wait()

	// Verify all companies were added
	companies := handler.CompaniesByUser("jim@example.org", func(privileges CompanyPrivileges) bool {
		return privileges.Admin
	})
	sort.Strings(companies)

	expected := make([]string, 10)
	for i := 0; i < 10; i++ {
		expected[i] = fmt.Sprintf("company-%d", i)
	}
	sort.Strings(expected)

	assert.Equal(t, expected, companies)
}

func TestPrivilegeHandler_Concurrent_Multiple_Operations(t *testing.T) {
	privileges := `
{
  "jim@example.org": {
    "initial-company": {
      "admin": true,
      "company": true,
      "consumer": false,
      "time": false,
      "invoicing": false,
      "accounting": false,
      "supplier": false,
      "salary": false
    }
  }
}`
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		_, _ = w.Write([]byte(privileges))
	}))
	defer server.Close()

	baseURL := server.Listener.Addr().String()
	handler := New(WithBaseURL(fmt.Sprintf("http://%s", baseURL)))

	var wg sync.WaitGroup

	// Fetch
	for i := 0; i < 5; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			_ = handler.Fetch()
		}()
	}

	// Process PrivilegeAdded
	for i := 0; i < 20; i++ {
		wg.Add(1)
		go func(idx int) {
			defer wg.Done()
			_ = handler.Process(&PrivilegeAdded{
				SequenceNo: 1,
				Email:      "jane@example.org",
				CompanyID:  fmt.Sprintf("company-%d", idx%5),
				Privilege:  PrivilegeCompany,
			})
		}(i)
	}

	// CompaniesByUser reads
	for i := 0; i < 50; i++ {
		wg.Add(1)
		email := "jim@example.org"
		if i%2 == 0 {
			email = "jane@example.org"
		}
		go func(e string) {
			defer wg.Done()
			_ = handler.CompaniesByUser(e, func(privileges CompanyPrivileges) bool {
				return privileges.Admin || privileges.Company
			})
		}(email)
	}

	// IsAllowed reads
	for i := 0; i < 50; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			_ = handler.IsAllowed("jim@example.org", "initial-company", func(privileges CompanyPrivileges) bool {
				return privileges.Admin
			})
		}()
	}

	wg.Wait()

	// Verify final state is consistent
	jimCompanies := handler.CompaniesByUser("jim@example.org", func(privileges CompanyPrivileges) bool {
		return privileges.Admin
	})
	assert.Contains(t, jimCompanies, "initial-company")

	janeCompanies := handler.CompaniesByUser("jane@example.org", func(privileges CompanyPrivileges) bool {
		return privileges.Company
	})
	sort.Strings(janeCompanies)

	expectedJane := []string{"company-0", "company-1", "company-2", "company-3", "company-4"}
	assert.Equal(t, expectedJane, janeCompanies)
}

func TestPrivilegeHandler_Setup(t *testing.T) {
	handler := New(WithBaseURL("base"))

	topology, err := goamqp.CollectTopology("some-service", handler.Setup()...)
	assert.NoError(t, err)
	wiring := map[string]string{}
	for _, e := range topology.Endpoints {
		assert.Equal(t, spec.DirectionConsume, e.Direction)
		assert.True(t, e.Ephemeral, "%s must be a per-replica consumer", e.RoutingKey)
		wiring[e.RoutingKey] = e.MessageType
	}
	// A key wired to the wrong type could turn a revocation into a grant.
	assert.Equal(t, map[string]string{
		"User.Added":        "client.UserAdded",
		"User.Removed":      "client.UserRemoved",
		"Privilege.Added":   "client.PrivilegeAdded",
		"Privilege.Removed": "client.PrivilegeRemoved",
	}, wiring)
}

func TestPrivilegeHandler_process(t *testing.T) {
	ctx := context.Background()
	handler := New(WithBaseURL("base"))
	admin := func(p CompanyPrivileges) bool { return p.Admin }

	assert.NoError(t, process[UserAdded](handler)(ctx, spec.ConsumableEvent[UserAdded]{
		Payload: UserAdded{SequenceNo: 1, Email: "jim@example.org", CompanyID: "abc-123"},
	}))
	assert.False(t, handler.IsAllowed("jim@example.org", "abc-123", admin))

	assert.NoError(t, process[PrivilegeAdded](handler)(ctx, spec.ConsumableEvent[PrivilegeAdded]{
		Payload: PrivilegeAdded{SequenceNo: 2, Email: "jim@example.org", CompanyID: "abc-123", Privilege: PrivilegeAdmin},
	}))
	assert.True(t, handler.IsAllowed("jim@example.org", "abc-123", admin))

	assert.NoError(t, process[PrivilegeRemoved](handler)(ctx, spec.ConsumableEvent[PrivilegeRemoved]{
		Payload: PrivilegeRemoved{SequenceNo: 3, Email: "jim@example.org", CompanyID: "abc-123", Privilege: PrivilegeAdmin},
	}))
	assert.False(t, handler.IsAllowed("jim@example.org", "abc-123", admin))

	assert.NoError(t, process[UserRemoved](handler)(ctx, spec.ConsumableEvent[UserRemoved]{
		Payload: UserRemoved{SequenceNo: 4, Email: "jim@example.org", CompanyID: "abc-123"},
	}))
	assert.Empty(t, handler.CompaniesByUser("jim@example.org", func(CompanyPrivileges) bool { return true }))
}

// privilegesOf returns the privileges of every membership in the handler.
func privilegesOf(h *PrivilegeHandler) map[string]map[string]CompanyPrivileges {
	h.RLock()
	defer h.RUnlock()
	result := map[string]map[string]CompanyPrivileges{}
	for email, companies := range h.state {
		for companyID, s := range companies {
			if !s.member {
				continue
			}
			if result[email] == nil {
				result[email] = map[string]CompanyPrivileges{}
			}
			result[email][companyID] = s.privileges
		}
	}
	return result
}

func TestPrivilegeHandler_Process_Orders_By_SequenceNo(t *testing.T) {
	const email, company = "jim@example.org", "abc-123"
	userAdded := func(seq int) any { return &UserAdded{Email: email, CompanyID: company, SequenceNo: seq} }
	userRemoved := func(seq int) any { return &UserRemoved{Email: email, CompanyID: company, SequenceNo: seq} }
	added := func(seq int) any {
		return &PrivilegeAdded{Email: email, CompanyID: company, Privilege: PrivilegeAdmin, SequenceNo: seq}
	}
	removed := func(seq int) any {
		return &PrivilegeRemoved{Email: email, CompanyID: company, Privilege: PrivilegeAdmin, SequenceNo: seq}
	}
	tests := []struct {
		name      string
		delivered []any
		member    bool
		admin     bool
	}{
		{name: "in order", delivered: []any{userAdded(1), added(2)}, member: true, admin: true},
		{name: "privilege before older user added keeps the grant", delivered: []any{added(2), userAdded(1)}, member: true, admin: true},
		{name: "late privilege added after user removed", delivered: []any{userAdded(1), userRemoved(3), added(2)}, member: false, admin: false},
		{name: "late privilege added after privilege removed", delivered: []any{userAdded(1), removed(3), added(2)}, member: true, admin: false},
		{name: "late user added does not revive privileges removed with the user", delivered: []any{userRemoved(3), added(2), userAdded(1)}, member: false, admin: false},
		{name: "re-added user starts without the removed privileges", delivered: []any{added(2), userRemoved(3), userAdded(4)}, member: true, admin: false},
		{name: "privilege granted after re-add", delivered: []any{userAdded(4), userRemoved(3), added(5), added(2)}, member: true, admin: true},
		{name: "late user removed does not remove a newer grant", delivered: []any{added(4), userRemoved(3)}, member: true, admin: true},
		{name: "late user removed does not remove a newer membership", delivered: []any{userAdded(4), userRemoved(3)}, member: true, admin: false},
		{name: "late privilege added before re-add stays revoked", delivered: []any{userAdded(1), userRemoved(3), added(2), userAdded(4)}, member: true, admin: false},
		{name: "privilege removed after user removed keeps the user out", delivered: []any{userRemoved(2), removed(3)}, member: false, admin: false},
		{name: "duplicate delivery is idempotent", delivered: []any{added(2), removed(3), added(2)}, member: true, admin: false},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			handler := New()
			for _, event := range tt.delivered {
				require.NoError(t, handler.Process(event))
			}
			all := func(CompanyPrivileges) bool { return true }
			assert.Equal(t, tt.member, handler.IsAllowed(email, company, all))
			assert.Equal(t, tt.admin, handler.IsAllowed(email, company, func(p CompanyPrivileges) bool { return p.Admin }))
			if tt.member {
				assert.Equal(t, []string{company}, handler.CompaniesByUser(email, all))
			} else {
				assert.Equal(t, []string{}, handler.CompaniesByUser(email, all))
			}
		})
	}
}

func TestPrivilegeHandler_Process_Without_SequenceNo_Fails_Closed(t *testing.T) {
	const email, company = "jim@example.org", "abc-123"
	admin := func(p CompanyPrivileges) bool { return p.Admin }
	handler := New()

	require.NoError(t, handler.Process(&PrivilegeAdded{Email: email, CompanyID: company, Privilege: PrivilegeAdmin}))
	require.NoError(t, handler.Process(&UserAdded{Email: email, CompanyID: company}))
	assert.False(t, handler.IsAllowed(email, company, func(CompanyPrivileges) bool { return true }), "an unordered addition is dropped")
	require.NoError(t, handler.Process(&UserAdded{Email: email, CompanyID: company, SequenceNo: 1}))
	assert.False(t, handler.IsAllowed(email, company, admin), "the dropped grant doesn't appear with a later membership")

	require.NoError(t, handler.Process(&PrivilegeAdded{Email: email, CompanyID: company, Privilege: PrivilegeAdmin, SequenceNo: 5}))
	require.NoError(t, handler.Process(&PrivilegeRemoved{Email: email, CompanyID: company, Privilege: PrivilegeAdmin}))
	assert.False(t, handler.IsAllowed(email, company, admin), "an unordered privilege removal applies")

	require.NoError(t, handler.Process(&PrivilegeAdded{Email: email, CompanyID: company, Privilege: PrivilegeTime, SequenceNo: 6}))
	require.NoError(t, handler.Process(&UserRemoved{Email: email, CompanyID: company}))
	assert.False(t, handler.IsAllowed(email, company, func(CompanyPrivileges) bool { return true }), "an unordered user removal applies")
}

// snapshotServer serves body as the /authz snapshot at seq; a negative seq sends no header.
func snapshotServer(t *testing.T, seq int, body string) *httptest.Server {
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if seq >= 0 {
			w.Header().Set(SequenceHeader, fmt.Sprint(seq))
		}
		_, _ = w.Write([]byte(body))
	}))
	t.Cleanup(server.Close)
	return server
}

func TestPrivilegeHandler_Fetch_Merges_Snapshot_By_SequenceNo(t *testing.T) {
	const email = "jim@example.org"
	admin := func(p CompanyPrivileges) bool { return p.Admin }
	all := func(CompanyPrivileges) bool { return true }
	snapshot := `{"jim@example.org": {"in-snapshot": {"admin": true}, "stale-in-snapshot": {"admin": true}}}`

	handler := New(WithBaseURL(snapshotServer(t, 10, snapshot).URL))
	// Delivered while the snapshot was read.
	require.NoError(t, handler.Process(&PrivilegeRemoved{Email: email, CompanyID: "in-snapshot", Privilege: PrivilegeAdmin, SequenceNo: 11}))
	require.NoError(t, handler.Process(&PrivilegeAdded{Email: email, CompanyID: "stale-in-snapshot", Privilege: PrivilegeTime, SequenceNo: 9}))
	require.NoError(t, handler.Process(&PrivilegeAdded{Email: email, CompanyID: "new", Privilege: PrivilegeAdmin, SequenceNo: 12}))
	require.NoError(t, handler.Process(&PrivilegeAdded{Email: email, CompanyID: "revoked", Privilege: PrivilegeAdmin, SequenceNo: 8}))
	require.NoError(t, handler.Process(&UserAdded{Email: email, CompanyID: "revoked-later", SequenceNo: 7}))
	require.NoError(t, handler.Process(&PrivilegeAdded{Email: email, CompanyID: "revoked-later", Privilege: PrivilegeAdmin, SequenceNo: 13}))

	require.NoError(t, handler.Fetch())

	assert.True(t, handler.IsAllowed(email, "in-snapshot", all))
	assert.False(t, handler.IsAllowed(email, "in-snapshot", admin), "a removal newer than the snapshot wins")
	assert.False(t, handler.IsAllowed(email, "stale-in-snapshot", func(p CompanyPrivileges) bool { return p.Time }), "the snapshot replaces an older fact")
	assert.True(t, handler.IsAllowed(email, "new", admin), "a grant newer than the snapshot is kept")
	assert.False(t, handler.IsAllowed(email, "revoked", all), "a pair missing from the snapshot is removed")
	assert.True(t, handler.IsAllowed(email, "revoked-later", admin), "a grant newer than the snapshot survives the removal")

	// Events the snapshot already covers are stale.
	require.NoError(t, handler.Process(&PrivilegeAdded{Email: email, CompanyID: "revoked", Privilege: PrivilegeAdmin, SequenceNo: 10}))
	assert.False(t, handler.IsAllowed(email, "revoked", all), "an event at the snapshot sequence is stale")
	require.NoError(t, handler.Process(&PrivilegeAdded{Email: email, CompanyID: "revoked", Privilege: PrivilegeAdmin, SequenceNo: 14}))
	assert.True(t, handler.IsAllowed(email, "revoked", admin))
}

func TestPrivilegeHandler_Fetch_Replaces_Pairs_Missing_From_Snapshot(t *testing.T) {
	handler := New(WithBaseURL(snapshotServer(t, 10, `{"jim@example.org": {"kept": {"company": true}}}`).URL))
	require.NoError(t, handler.Process(&PrivilegeAdded{Email: "jim@example.org", CompanyID: "gone", Privilege: PrivilegeAdmin, SequenceNo: 3}))
	require.NoError(t, handler.Process(&PrivilegeAdded{Email: "jane@example.org", CompanyID: "gone", Privilege: PrivilegeAdmin, SequenceNo: 4}))

	require.NoError(t, handler.Fetch())

	assert.Equal(t, map[string]map[string]CompanyPrivileges{"jim@example.org": {"kept": {Company: true}}}, privilegesOf(handler))
	handler.RLock()
	defer handler.RUnlock()
	assert.NotContains(t, handler.state, "jane@example.org", "an email without companies is dropped")
}

func TestPrivilegeHandler_Fetch_Without_SequenceHeader_Keeps_Sequenced_Facts(t *testing.T) {
	handler := New(WithBaseURL(snapshotServer(t, -1, `{"jim@example.org": {"abc-123": {"admin": true}}}`).URL))
	require.NoError(t, handler.Process(&PrivilegeRemoved{Email: "jim@example.org", CompanyID: "abc-123", Privilege: PrivilegeAdmin, SequenceNo: 3}))
	require.NoError(t, handler.Process(&PrivilegeAdded{Email: "jim@example.org", CompanyID: "other", Privilege: PrivilegeAdmin, SequenceNo: 4}))

	require.NoError(t, handler.Fetch())

	assert.Equal(t, map[string]map[string]CompanyPrivileges{
		"jim@example.org": {"abc-123": {}, "other": {Admin: true}},
	}, privilegesOf(handler))
}

func TestPrivilegeHandler_Fetch_Rejects_Bad_Responses(t *testing.T) {
	tests := []struct {
		name   string
		status int
		header string
		err    string
	}{
		{name: "status", status: http.StatusUnauthorized, err: "fetch privileges: unexpected status 401 Unauthorized"},
		{name: "invalid header", status: http.StatusOK, header: "abc", err: `fetch privileges: invalid X-Authz-Sequence header "abc"`},
		{name: "negative header", status: http.StatusOK, header: "-1", err: `fetch privileges: invalid X-Authz-Sequence header "-1"`},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
				if tt.header != "" {
					w.Header().Set(SequenceHeader, tt.header)
				}
				w.WriteHeader(tt.status)
				_, _ = w.Write([]byte(`{"jim@example.org": {"abc-123": {"admin": true}}}`))
			}))
			defer server.Close()
			handler := New(WithBaseURL(server.URL))
			require.NoError(t, handler.Process(&PrivilegeAdded{Email: "jane@example.org", CompanyID: "abc-123", Privilege: PrivilegeTime, SequenceNo: 1}))

			assert.EqualError(t, handler.Fetch(), tt.err)
			assert.Equal(t, map[string]map[string]CompanyPrivileges{"jane@example.org": {"abc-123": {Time: true}}}, privilegesOf(handler), "state is untouched")
		})
	}
}

func TestPrivilegeHandler_Revocation_During_Fetch_Wins(t *testing.T) {
	// The consumers are bound before Fetch, so a revocation published after authz-service
	// read the snapshot is processed while the response is still on its way.
	handler := New()
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		// assert, not require: this runs on the server's goroutine.
		assert.NoError(t, handler.Process(&UserRemoved{Email: "jim@example.org", CompanyID: "abc-123", SequenceNo: 21}))
		w.Header().Set(SequenceHeader, "20")
		_, _ = w.Write([]byte(`{"jim@example.org": {"abc-123": {"admin": true}}}`))
	}))
	defer server.Close()
	handler.baseURL = server.URL

	require.NoError(t, handler.Fetch())

	assert.False(t, handler.IsAllowed("jim@example.org", "abc-123", func(CompanyPrivileges) bool { return true }))
}

func TestPrivilegeHandler_Fetch_Ignores_Snapshot_Older_Than_Merged_One(t *testing.T) {
	const email = "jim@example.org"
	all := func(CompanyPrivileges) bool { return true }
	handler := New(WithBaseURL(snapshotServer(t, 12, `{}`).URL))
	require.NoError(t, handler.Process(&PrivilegeAdded{Email: email, CompanyID: "abc-123", Privilege: PrivilegeAdmin, SequenceNo: 5}))
	require.NoError(t, handler.Process(&UserRemoved{Email: email, CompanyID: "abc-123", SequenceNo: 11}))
	require.NoError(t, handler.Fetch())

	// A concurrent Fetch read at seq 10, before the removal, merges last.
	handler.baseURL = snapshotServer(t, 10, `{"jim@example.org": {"abc-123": {"admin": true}}}`).URL
	require.NoError(t, handler.Fetch())
	assert.False(t, handler.IsAllowed(email, "abc-123", all))

	// So does a snapshot without a sequence number.
	handler.baseURL = snapshotServer(t, -1, `{"jim@example.org": {"abc-123": {"admin": true}}}`).URL
	require.NoError(t, handler.Fetch())
	assert.False(t, handler.IsAllowed(email, "abc-123", all))
}

func TestPrivilegeHandler_Unordered_Removal_Holds_Until_Next_Snapshot(t *testing.T) {
	const email, company = "jim@example.org", "abc-123"
	all := func(CompanyPrivileges) bool { return true }
	admin := func(p CompanyPrivileges) bool { return p.Admin }
	handler := New(WithBaseURL(snapshotServer(t, 30, `{"jim@example.org": {"abc-123": {"admin": true}, "def-456": {"time": true}}}`).URL))
	require.NoError(t, handler.Process(&UserAdded{Email: email, CompanyID: company, SequenceNo: 10}))
	require.NoError(t, handler.Process(&PrivilegeAdded{Email: email, CompanyID: company, Privilege: PrivilegeAdmin, SequenceNo: 11}))
	require.NoError(t, handler.Process(&PrivilegeAdded{Email: email, CompanyID: "def-456", Privilege: PrivilegeTime, SequenceNo: 12}))

	require.NoError(t, handler.Process(&UserRemoved{Email: email, CompanyID: company}))
	require.NoError(t, handler.Process(&PrivilegeRemoved{Email: email, CompanyID: "def-456", Privilege: PrivilegeTime}))
	require.NoError(t, handler.Process(&PrivilegeAdded{Email: email, CompanyID: company, Privilege: PrivilegeSalary, SequenceNo: 15}))
	require.NoError(t, handler.Process(&PrivilegeAdded{Email: email, CompanyID: company, Privilege: PrivilegeAdmin, SequenceNo: 16}))
	require.NoError(t, handler.Process(&UserAdded{Email: email, CompanyID: company, SequenceNo: 17}))
	require.NoError(t, handler.Process(&PrivilegeAdded{Email: email, CompanyID: "def-456", Privilege: PrivilegeTime, SequenceNo: 18}))
	assert.False(t, handler.IsAllowed(email, company, all), "no event overrides an unordered user removal")
	assert.False(t, handler.IsAllowed(email, "def-456", func(p CompanyPrivileges) bool { return p.Time }), "no event overrides an unordered privilege removal")

	require.NoError(t, handler.Fetch())
	assert.True(t, handler.IsAllowed(email, company, admin), "the next snapshot replaces the unordered removal")
	assert.True(t, handler.IsAllowed(email, "def-456", func(p CompanyPrivileges) bool { return p.Time }))
}

func TestPrivilegeHandler_Fetch_Deletes_Unordered_Removal_Missing_From_Snapshot(t *testing.T) {
	handler := New(WithBaseURL(snapshotServer(t, 30, `{}`).URL))
	require.NoError(t, handler.Process(&UserRemoved{Email: "jim@example.org", CompanyID: "abc-123"}))

	require.NoError(t, handler.Fetch())

	handler.RLock()
	defer handler.RUnlock()
	assert.Empty(t, handler.state)
}

func TestPrivilegeHandler_Process_Unknown_Privilege_Creates_No_State(t *testing.T) {
	handler := New()
	require.NoError(t, handler.Process(&PrivilegeRemoved{Email: "jim@example.org", CompanyID: "abc-123", Privilege: "UNKNOWN", SequenceNo: 1}))
	require.NoError(t, handler.Process(&PrivilegeAdded{Email: "jim@example.org", CompanyID: "abc-123", Privilege: "UNKNOWN", SequenceNo: 2}))

	handler.RLock()
	defer handler.RUnlock()
	assert.Empty(t, handler.state)
}

func TestPrivilegeHandler_Fetch_Retries_While_ReadView_Not_Ready(t *testing.T) {
	var calls atomic.Int32
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if calls.Add(1) < 3 {
			w.WriteHeader(http.StatusServiceUnavailable)
			return
		}
		w.Header().Set(SequenceHeader, "7")
		_, _ = w.Write([]byte(`{"jim@example.org": {"abc-123": {"admin": true}}}`))
	}))
	defer server.Close()
	handler := New(WithBaseURL(server.URL))
	handler.retryDelay = time.Millisecond

	require.NoError(t, handler.Fetch())

	assert.Equal(t, int32(3), calls.Load())
	assert.True(t, handler.IsAllowed("jim@example.org", "abc-123", func(p CompanyPrivileges) bool { return p.Admin }))
}

func TestPrivilegeHandler_Fetch_Gives_Up_When_ReadView_Stays_Behind(t *testing.T) {
	var calls atomic.Int32
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		calls.Add(1)
		w.WriteHeader(http.StatusServiceUnavailable)
	}))
	defer server.Close()
	handler := New(WithBaseURL(server.URL))
	handler.retryDelay = time.Millisecond
	handler.retries = 2

	assert.EqualError(t, handler.Fetch(), "fetch privileges: authz-service read view not ready")
	assert.Equal(t, int32(3), calls.Load())
}

func TestPrivilegeHandler_Process_Drops_Invalid_SequenceNo(t *testing.T) {
	for _, seq := range []int{-1, math.MaxInt} {
		handler := New()
		require.NoError(t, handler.Process(&PrivilegeAdded{Email: "jim@example.org", CompanyID: "abc-123", Privilege: PrivilegeAdmin, SequenceNo: seq}))
		require.NoError(t, handler.Process(&UserRemoved{Email: "jim@example.org", CompanyID: "abc-123", SequenceNo: seq}))
		handler.RLock()
		assert.Empty(t, handler.state, "seq %d", seq)
		handler.RUnlock()
	}
}