diff --git a/CLAUDE.md b/CLAUDE.md index 3fb784e..5ea8b8c 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -23,10 +23,9 @@ import client "gitea.unbound.se/shiny/authz_client" handler := client.New(client.WithBaseURL("http://authz-service")) // Check user privileges -privileges := handler.Get(email, companyID) -if privileges.Invoicing { - // User has invoicing privileges -} +allowed := handler.IsAllowed(email, companyID, func(p client.CompanyPrivileges) bool { + return p.Invoicing +}) ``` ### Privileges @@ -43,4 +42,10 @@ 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. 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. +Registers per-replica (transient) go-messaging-amqp consumers for privilege events from authz-service (`Setup()`). Each routing key gets its own queue, so events arrive in any order. `Process` orders them by the `sequenceNo` authz-service's event store stamps on every event: all four events come from authz-service's `Company` aggregate, so the global sequence number orders them within a company. Each (email, company) keeps the sequence number of the last fact about membership and about each privilege; an event only overrides older facts. `User.Removed` stamps every privilege, so a late grant can't come back. An event without `sequenceNo` fails closed: additions are dropped, and removals apply and block every later event for those facts until the next snapshot. An event with a negative or implausibly large `sequenceNo` is dropped. Tests that seed the handler must set `SequenceNo`. + +### Startup + +Call `Fetch()` **after** `conn.Start` (consumers bound), never before: events published between the snapshot and the binding would otherwise be lost. authz-service only serves a snapshot whose read view has applied every stored event. It sends that position in the `X-Authz-Sequence` header and answers 503 while the read view lags (backlog, backfill, reset); `Fetch` retries 503 for about a minute. `Fetch` merges the snapshot as facts at that sequence number (newer event facts win, pairs missing from the snapshot are removed) and drops later events at or below it. A snapshot older than one already merged is ignored, so concurrent or repeated fetches are safe. A snapshot without the header merges at 0 and logs a warning (rollout window only). + +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. diff --git a/client.go b/client.go index a506cec..81e8b2b 100644 --- a/client.go +++ b/client.go @@ -3,16 +3,29 @@ package client import ( "context" "encoding/json" + "errors" "fmt" "io" + "log/slog" + "math" "net/http" "reflect" + "strconv" "sync" + "time" goamqp "codeberg.org/messaging/go-messaging-amqp" spec "codeberg.org/messaging/messaging" ) +// SequenceHeader carries the global sequence number of the authz-service read view the +// /authz snapshot was read at. +const SequenceHeader = "X-Authz-Sequence" + +// pending marks a fact cleared by an event without a sequence number. No event can +// override it; only a snapshot can. +const pending = math.MaxInt + // CompanyPrivileges contains the privileges for a combination of email address and company id type CompanyPrivileges struct { Admin bool `json:"admin"` @@ -25,13 +38,84 @@ type CompanyPrivileges struct { Salary bool `json:"salary"` } +// field returns the flag for privilege, or nil for an unknown privilege. +func (c *CompanyPrivileges) field(privilege Privilege) *bool { + switch privilege { + case PrivilegeAdmin: + return &c.Admin + case PrivilegeCompany: + return &c.Company + case PrivilegeConsumer: + return &c.Consumer + case PrivilegeTime: + return &c.Time + case PrivilegeInvoicing: + return &c.Invoicing + case PrivilegeAccounting: + return &c.Accounting + case PrivilegeSupplier: + return &c.Supplier + case PrivilegeSalary: + return &c.Salary + } + return nil +} + +// companyState is what the handler knows about one email in one company. Every +// privilege event for a company comes from authz-service's Company aggregate, so +// its global sequence number orders the events for the pair. The four routing +// keys arrive on separate queues in any order; a fact is only overwritten by a +// fact with a higher sequence number. +type companyState struct { + privileges CompanyPrivileges + // privilegeSeq is the sequence number of the last fact about each privilege. + privilegeSeq map[Privilege]int + // member is whether the user belongs to the company, as of memberSeq. A + // User.Removed stamps every privilege with its sequence number, so a grant + // older than the removal can't come back. + member bool + memberSeq int +} + +func newCompanyState() *companyState { + return &companyState{privilegeSeq: map[Privilege]int{}} +} + +// clearPending turns facts cleared without a sequence number into unordered facts, so +// the snapshot being merged replaces them. +func (s *companyState) clearPending() { + if s.memberSeq == pending { + s.memberSeq = 0 + } + for p, seq := range s.privilegeSeq { + if seq == pending { + s.privilegeSeq[p] = 0 + } + } +} + +// maxSeq is the highest sequence number of any fact in the state. +func (s *companyState) maxSeq() int { + m := s.memberSeq + for _, seq := range s.privilegeSeq { + m = max(m, seq) + } + return m +} + // PrivilegeHandler processes PrivilegeAdded-events and fetches the initial set of privileges from an authz-service type PrivilegeHandler struct { *sync.RWMutex - client *http.Client - baseURL string - apiKey string - privileges map[string]map[string]*CompanyPrivileges + client *http.Client + baseURL string + apiKey string + state map[string]map[string]*companyState + // floor is the sequence number of the newest snapshot. Its effects are all in + // the state, so an event at or below it is stale. + floor int + // retries is how many times Fetch retries a snapshot authz-service isn't ready to serve. + retries int + retryDelay time.Duration } // OptsFunc is used to configure the PrivilegeHandler @@ -55,9 +139,11 @@ func WithAPIKey(key string) OptsFunc { func New(opts ...OptsFunc) *PrivilegeHandler { handler := &PrivilegeHandler{ RWMutex: &sync.RWMutex{}, - client: &http.Client{}, + client: &http.Client{Timeout: 30 * time.Second}, baseURL: "http://authz-service", - privileges: map[string]map[string]*CompanyPrivileges{}, + state: map[string]map[string]*companyState{}, + retries: 60, + retryDelay: time.Second, } for _, opt := range opts { opt(handler) @@ -65,8 +151,27 @@ func New(opts ...OptsFunc) *PrivilegeHandler { return handler } -// Fetch the initial set of privileges from an authz-service +// Fetch a snapshot of all privileges from an authz-service and merge it into the state. +// +// Call it after the AMQP connection has started: authz-service only serves a snapshot +// that includes every stored event, so together with the bound queues no event is +// missed. While its read view is behind it answers 503, and Fetch retries. The snapshot +// is applied as facts at its sequence number: a fact from an event newer than the +// snapshot is kept, everything else is replaced, and a pair missing from the snapshot is +// removed. A snapshot older than one already merged is ignored. func (h *PrivilegeHandler) Fetch() error { + for attempt := 0; ; attempt++ { + err := h.fetch() + if !errors.Is(err, errNotReady) || attempt >= h.retries { + return err + } + time.Sleep(h.retryDelay) + } +} + +var errNotReady = errors.New("fetch privileges: authz-service read view not ready") + +func (h *PrivilegeHandler) fetch() error { req, err := http.NewRequest(http.MethodGet, fmt.Sprintf("%s/authz", h.baseURL), nil) if err != nil { return err @@ -80,21 +185,83 @@ func (h *PrivilegeHandler) Fetch() error { if err != nil { return err } + defer func() { _ = resp.Body.Close() }() + + if resp.StatusCode == http.StatusServiceUnavailable { + return errNotReady + } + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("fetch privileges: unexpected status %s", resp.Status) + } buff, err := io.ReadAll(resp.Body) if err != nil { return err } - h.Lock() - defer h.Unlock() - err = json.Unmarshal(buff, &h.privileges) - if err != nil { + var snapshot map[string]map[string]CompanyPrivileges + if err := json.Unmarshal(buff, &snapshot); err != nil { return err } + + seq := 0 + if v := resp.Header.Get(SequenceHeader); v != "" { + seq, err = strconv.Atoi(v) + if err != nil || seq < 0 { + return fmt.Errorf("fetch privileges: invalid %s header %q", SequenceHeader, v) + } + } else { + // ponytail: only during the rollout window, before authz-service sends the header. + slog.Warn("authz snapshot has no sequence number; events older than it can revert it", "header", SequenceHeader) + } + + h.Lock() + defer h.Unlock() + h.merge(snapshot, seq) return nil } +func (h *PrivilegeHandler) merge(snapshot map[string]map[string]CompanyPrivileges, seq int) { + if seq < h.floor { + // A newer snapshot is already merged; this one could only bring back removed state. + slog.Warn("ignoring authz snapshot older than the one already merged", "seq", seq, "floor", h.floor) + return + } + for email, companies := range h.state { + for companyID, s := range companies { + s.clearPending() + if _, exists := snapshot[email][companyID]; exists { + continue + } + if s.maxSeq() <= seq { + // Nothing newer than the snapshot; the floor keeps stale events out. + delete(companies, companyID) + continue + } + removeUser(s, seq, false) + } + if len(companies) == 0 { + delete(h.state, email) + } + } + for email, companies := range snapshot { + for companyID, privileges := range companies { + s := h.company(email, companyID) + if s.memberSeq <= seq { + s.member = true + s.memberSeq = seq + } + for _, p := range AllPrivilege { + if s.privilegeSeq[p] <= seq { + *s.privileges.field(p) = *privileges.field(p) + s.privilegeSeq[p] = seq + } + } + } + } + h.floor = max(h.floor, seq) +} + func (h *PrivilegeHandler) Setup() []goamqp.Setup { return []goamqp.Setup{ goamqp.TransientEventStreamConsumer("User.Added", process[UserAdded](h)), @@ -116,71 +283,118 @@ func process[T privilegeEvent](h *PrivilegeHandler) spec.EventHandler[T] { } } -// Process privilege-related events and update the internal state +// Process privilege-related events and update the internal state. +// +// Events are applied by sequence number, not by arrival order. An event without a +// sequence number can't be ordered, so it fails closed: a removal is applied and an +// addition is dropped. func (h *PrivilegeHandler) Process(msg any) error { h.Lock() defer h.Unlock() switch ev := msg.(type) { case *UserAdded: - // 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 h.stale(ev.SequenceNo, true, ev) { + return nil } - if _, exists := priv[ev.CompanyID]; !exists { - priv[ev.CompanyID] = &CompanyPrivileges{} + s := h.company(ev.Email, ev.CompanyID) + if ev.SequenceNo > s.memberSeq { + s.member = true + s.memberSeq = ev.SequenceNo } return nil case *UserRemoved: - if priv, exists := h.privileges[ev.Email]; exists { - delete(priv, ev.CompanyID) + if h.stale(ev.SequenceNo, false, ev) { + return nil } + removeUser(h.company(ev.Email, ev.CompanyID), ev.SequenceNo, ev.SequenceNo == 0) return nil case *PrivilegeAdded: - h.setPrivileges(ev.Email, ev.CompanyID, ev.Privilege, true) + if h.stale(ev.SequenceNo, true, ev) { + return nil + } + h.setPrivilege(ev.Email, ev.CompanyID, ev.Privilege, ev.SequenceNo, true) return nil case *PrivilegeRemoved: - h.setPrivileges(ev.Email, ev.CompanyID, ev.Privilege, false) + if h.stale(ev.SequenceNo, false, ev) { + return nil + } + h.setPrivilege(ev.Email, ev.CompanyID, ev.Privilege, ev.SequenceNo, false) return nil default: - fmt.Printf("Got unexpected message type (%s): '%+v'\n", reflect.TypeOf(msg).String(), msg) + slog.Error("unexpected privilege message type", "type", reflect.TypeOf(msg).String()) return fmt.Errorf("unexpected event type: '%s'", reflect.TypeOf(msg)) } } -func (h *PrivilegeHandler) setPrivileges(email, companyId string, privilege Privilege, set bool) { - if priv, exists := h.privileges[email]; exists { - if c, exists := priv[companyId]; exists { - switch privilege { - case PrivilegeAdmin: - c.Admin = set - case PrivilegeCompany: - c.Company = set - case PrivilegeConsumer: - c.Consumer = set - case PrivilegeTime: - c.Time = set - case PrivilegeInvoicing: - c.Invoicing = set - case PrivilegeAccounting: - c.Accounting = set - case PrivilegeSupplier: - c.Supplier = set - case PrivilegeSalary: - c.Salary = set - } - } else { - priv[companyId] = &CompanyPrivileges{} - h.setPrivileges(email, companyId, privilege, set) +// stale reports whether an event must be skipped: an addition without a sequence +// number, or any event already covered by the latest snapshot. +func (h *PrivilegeHandler) stale(seq int, addition bool, ev any) bool { + if seq < 0 || seq >= pending { + slog.Error("dropping privilege event with invalid sequence number", "type", reflect.TypeOf(ev).String(), "seq", seq) + return true + } + if seq == 0 { + if addition { + slog.Warn("dropping privilege addition without sequence number", "type", reflect.TypeOf(ev).String()) } - } else { - h.privileges[email] = map[string]*CompanyPrivileges{} - h.setPrivileges(email, companyId, privilege, set) + return addition + } + return seq <= h.floor +} + +func (h *PrivilegeHandler) company(email, companyID string) *companyState { + companies, exists := h.state[email] + if !exists { + companies = map[string]*companyState{} + h.state[email] = companies + } + s, exists := companies[companyID] + if !exists { + s = newCompanyState() + companies[companyID] = s + } + return s +} + +// removeUser removes the user at seq, overriding every fact older than seq. An +// unordered removal (all) overrides every fact and marks them pending, so no event can +// bring them back before the next snapshot. +func removeUser(s *companyState, seq int, all bool) { + stamp := seq + if all { + stamp = pending + } + if all || s.memberSeq < seq { + s.member = false + s.memberSeq = max(s.memberSeq, stamp) + } + for _, p := range AllPrivilege { + if all || s.privilegeSeq[p] < seq { + *s.privileges.field(p) = false + s.privilegeSeq[p] = max(s.privilegeSeq[p], stamp) + } + } +} + +func (h *PrivilegeHandler) setPrivilege(email, companyID string, privilege Privilege, seq int, set bool) { + if !privilege.IsValid() { + return + } + s := h.company(email, companyID) + if seq != 0 && seq <= s.privilegeSeq[privilege] { + return + } + *s.privileges.field(privilege) = set + if seq == 0 { + // Only an unordered removal gets here; keep it until the next snapshot. + seq = pending + } + s.privilegeSeq[privilege] = max(s.privilegeSeq[privilege], seq) + // authz-service's aggregate adds the user when a privilege is granted. + if set && seq > s.memberSeq { + s.member = true + s.memberSeq = seq } } @@ -188,12 +402,10 @@ func (h *PrivilegeHandler) setPrivileges(email, companyId string, privilege Priv func (h *PrivilegeHandler) CompaniesByUser(email string, predicate func(privileges CompanyPrivileges) bool) []string { h.RLock() defer h.RUnlock() - var result []string - if p, exists := h.privileges[email]; exists { - for k, v := range p { - if predicate(*v) { - result = append(result, k) - } + result := []string{} + for k, s := range h.state[email] { + if s.member && predicate(s.privileges) { + result = append(result, k) } } return result @@ -203,11 +415,8 @@ func (h *PrivilegeHandler) CompaniesByUser(email string, predicate func(privileg func (h *PrivilegeHandler) IsAllowed(email, companyID string, predicate func(privileges CompanyPrivileges) bool) bool { h.RLock() defer h.RUnlock() - if p, exists := h.privileges[email]; exists { - if v, exists := p[companyID]; exists { - return predicate(*v) - } + if s, exists := h.state[email][companyID]; exists && s.member { + return predicate(s.privileges) } - return false } diff --git a/client_test.go b/client_test.go index eaea615..cfdc4da 100644 --- a/client_test.go +++ b/client_test.go @@ -3,11 +3,14 @@ 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" @@ -27,9 +30,10 @@ func TestPrivilegeHandler_Process_PrivilegeRemoved(t *testing.T) { handler := New(WithBaseURL("base")) err := handler.Process(&PrivilegeAdded{ - Email: "jim@example.org", - CompanyID: "abc-123", - Privilege: PrivilegeAdmin, + SequenceNo: 1, + Email: "jim@example.org", + CompanyID: "abc-123", + Privilege: PrivilegeAdmin, }) assert.NoError(t, err) @@ -40,9 +44,10 @@ func TestPrivilegeHandler_Process_PrivilegeRemoved(t *testing.T) { assert.Equal(t, []string{"abc-123"}, companies) err = handler.Process(&PrivilegeRemoved{ - Email: "jim@example.org", - CompanyID: "abc-123", - Privilege: PrivilegeAdmin, + SequenceNo: 2, + Email: "jim@example.org", + CompanyID: "abc-123", + Privilege: PrivilegeAdmin, }) assert.NoError(t, err) @@ -57,14 +62,16 @@ func TestPrivilegeHandler_Process_UserAdded_And_UserRemoved(t *testing.T) { handler := New(WithBaseURL("base")) err := handler.Process(&UserAdded{ - Email: "jim@example.org", - CompanyID: "abc-123", + SequenceNo: 1, + Email: "jim@example.org", + CompanyID: "abc-123", }) assert.NoError(t, err) err = handler.Process(&UserAdded{ - Email: "jim@example.org", - CompanyID: "abc-456", + SequenceNo: 2, + Email: "jim@example.org", + CompanyID: "abc-456", }) assert.NoError(t, err) @@ -75,14 +82,16 @@ func TestPrivilegeHandler_Process_UserAdded_And_UserRemoved(t *testing.T) { assert.Equal(t, []string{"abc-123", "abc-456"}, companies) err = handler.Process(&UserRemoved{ - Email: "jim@example.org", - CompanyID: "abc-123", + SequenceNo: 3, + Email: "jim@example.org", + CompanyID: "abc-123", }) assert.NoError(t, err) err = handler.Process(&UserRemoved{ - Email: "jim@example.org", - CompanyID: "abc-456", + SequenceNo: 4, + Email: "jim@example.org", + CompanyID: "abc-456", }) assert.NoError(t, err) @@ -96,8 +105,8 @@ func TestPrivilegeHandler_Process_UserAdded_Keeps_Existing_Privileges(t *testing // 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} + 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}, @@ -118,14 +127,15 @@ func TestPrivilegeHandler_Process_UserAdded_Keeps_Existing_Privileges(t *testing 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(&UserAdded{SequenceNo: 1, Email: "jim@example.org", CompanyID: "abc-123"})) assert.NoError(t, handler.Process(&PrivilegeAdded{ - Email: "jim@example.org", - CompanyID: "abc-123", - Privilege: PrivilegeAdmin, + SequenceNo: 2, + 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"})) + 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 { @@ -150,8 +160,9 @@ func TestPrivilegeHandler_GetCompanies_No_Companies_Found(t *testing.T) { handler := New(WithBaseURL("base")) err := handler.Process(&UserAdded{ - Email: "jim@example.org", - CompanyID: "abc-123", + SequenceNo: 1, + Email: "jim@example.org", + CompanyID: "abc-123", }) assert.NoError(t, err) @@ -168,8 +179,9 @@ func TestPrivilegeHandler_GetCompanies_No_Companies_Found(t *testing.T) { assert.Equal(t, []string{"abc-123"}, companies) err = handler.Process(&UserRemoved{ - Email: "jim@example.org", - CompanyID: "abc-123", + SequenceNo: 2, + Email: "jim@example.org", + CompanyID: "abc-123", }) assert.NoError(t, err) @@ -183,9 +195,10 @@ func TestPrivilegeHandler_GetCompanies_Company_With_Company_Access_Found(t *test handler := New(WithBaseURL("base")) err := handler.Process(&PrivilegeAdded{ - Email: "jim@example.org", - CompanyID: "abc-123", - Privilege: PrivilegeCompany, + SequenceNo: 1, + Email: "jim@example.org", + CompanyID: "abc-123", + Privilege: PrivilegeCompany, }) assert.NoError(t, err) @@ -200,9 +213,10 @@ func TestPrivilegeHandler_GetCompanies_Company_With_Admin_Access_Found(t *testin handler := New(WithBaseURL("base")) err := handler.Process(&PrivilegeAdded{ - Email: "jim@example.org", - CompanyID: "abc-123", - Privilege: PrivilegeConsumer, + SequenceNo: 1, + Email: "jim@example.org", + CompanyID: "abc-123", + Privilege: PrivilegeConsumer, }) assert.NoError(t, err) @@ -227,9 +241,10 @@ func TestPrivilegeHandler_IsAllowed_Return_True_If_Privilege_Exists(t *testing.T handler := New(WithBaseURL("base")) _ = handler.Process(&PrivilegeAdded{ - Email: "jim@example.org", - CompanyID: "abc-123", - Privilege: PrivilegeTime, + SequenceNo: 1, + Email: "jim@example.org", + CompanyID: "abc-123", + Privilege: PrivilegeTime, }) result := handler.IsAllowed("jim@example.org", "abc-123", func(privileges CompanyPrivileges) bool { @@ -239,9 +254,10 @@ func TestPrivilegeHandler_IsAllowed_Return_True_If_Privilege_Exists(t *testing.T assert.True(t, result) _ = handler.Process(&PrivilegeAdded{ - Email: "jim@example.org", - CompanyID: "abc-123", - Privilege: PrivilegeInvoicing, + SequenceNo: 2, + Email: "jim@example.org", + CompanyID: "abc-123", + Privilege: PrivilegeInvoicing, }) result = handler.IsAllowed("jim@example.org", "abc-123", func(privileges CompanyPrivileges) bool { @@ -251,9 +267,10 @@ func TestPrivilegeHandler_IsAllowed_Return_True_If_Privilege_Exists(t *testing.T assert.True(t, result) _ = handler.Process(&PrivilegeAdded{ - Email: "jim@example.org", - CompanyID: "abc-123", - Privilege: PrivilegeAccounting, + SequenceNo: 3, + Email: "jim@example.org", + CompanyID: "abc-123", + Privilege: PrivilegeAccounting, }) result = handler.IsAllowed("jim@example.org", "abc-123", func(privileges CompanyPrivileges) bool { @@ -263,9 +280,10 @@ func TestPrivilegeHandler_IsAllowed_Return_True_If_Privilege_Exists(t *testing.T assert.True(t, result) _ = handler.Process(&PrivilegeAdded{ - Email: "jim@example.org", - CompanyID: "abc-123", - Privilege: PrivilegeSupplier, + SequenceNo: 4, + Email: "jim@example.org", + CompanyID: "abc-123", + Privilege: PrivilegeSupplier, }) result = handler.IsAllowed("jim@example.org", "abc-123", func(privileges CompanyPrivileges) bool { @@ -275,9 +293,10 @@ func TestPrivilegeHandler_IsAllowed_Return_True_If_Privilege_Exists(t *testing.T assert.True(t, result) _ = handler.Process(&PrivilegeAdded{ - Email: "jim@example.org", - CompanyID: "abc-123", - Privilege: PrivilegeSalary, + SequenceNo: 5, + Email: "jim@example.org", + CompanyID: "abc-123", + Privilege: PrivilegeSalary, }) result = handler.IsAllowed("jim@example.org", "abc-123", func(privileges CompanyPrivileges) bool { @@ -386,7 +405,7 @@ func TestPrivilegeHandler_Fetch_Valid(t *testing.T) { err := handler.Fetch() assert.NoError(t, err) - expectedPrivileges := map[string]map[string]*CompanyPrivileges{ + expectedPrivileges := map[string]map[string]CompanyPrivileges{ "jim@example.org": { "00010203-0405-4607-8809-0a0b0c0d0e0f": { Admin: false, @@ -400,7 +419,7 @@ func TestPrivilegeHandler_Fetch_Valid(t *testing.T) { }, }, } - assert.Equal(t, expectedPrivileges, handler.privileges) + assert.Equal(t, expectedPrivileges, privilegesOf(handler)) } func TestPrivilegeHandler_Fetch_Concurrent_Fetches(t *testing.T) { @@ -450,7 +469,7 @@ func TestPrivilegeHandler_Fetch_Concurrent_Fetches(t *testing.T) { } // Verify privileges were set correctly - expectedPrivileges := map[string]map[string]*CompanyPrivileges{ + expectedPrivileges := map[string]map[string]CompanyPrivileges{ "jim@example.org": { "00010203-0405-4607-8809-0a0b0c0d0e0f": { Admin: false, @@ -464,7 +483,7 @@ func TestPrivilegeHandler_Fetch_Concurrent_Fetches(t *testing.T) { }, }, } - assert.Equal(t, expectedPrivileges, handler.privileges) + assert.Equal(t, expectedPrivileges, privilegesOf(handler)) } func TestPrivilegeHandler_Concurrent_Fetch_And_Read(t *testing.T) { @@ -559,9 +578,10 @@ func TestPrivilegeHandler_Concurrent_Process_And_Read(t *testing.T) { go func(id string) { defer wg.Done() _ = handler.Process(&PrivilegeAdded{ - Email: "jim@example.org", - CompanyID: id, - Privilege: PrivilegeAdmin, + SequenceNo: 1, + Email: "jim@example.org", + CompanyID: id, + Privilege: PrivilegeAdmin, }) }(companyID) } @@ -635,9 +655,10 @@ func TestPrivilegeHandler_Concurrent_Multiple_Operations(t *testing.T) { go func(idx int) { defer wg.Done() _ = handler.Process(&PrivilegeAdded{ - Email: "jane@example.org", - CompanyID: fmt.Sprintf("company-%d", idx%5), - Privilege: PrivilegeCompany, + SequenceNo: 1, + Email: "jane@example.org", + CompanyID: fmt.Sprintf("company-%d", idx%5), + Privilege: PrivilegeCompany, }) }(i) } @@ -711,22 +732,333 @@ func TestPrivilegeHandler_process(t *testing.T) { admin := func(p CompanyPrivileges) bool { return p.Admin } assert.NoError(t, process[UserAdded](handler)(ctx, spec.ConsumableEvent[UserAdded]{ - Payload: UserAdded{Email: "jim@example.org", CompanyID: "abc-123"}, + 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{Email: "jim@example.org", CompanyID: "abc-123", Privilege: PrivilegeAdmin}, + 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{Email: "jim@example.org", CompanyID: "abc-123", Privilege: PrivilegeAdmin}, + 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{Email: "jim@example.org", CompanyID: "abc-123"}, + 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() + } +} diff --git a/events.go b/events.go index 7241616..cb7847d 100644 --- a/events.go +++ b/events.go @@ -1,15 +1,18 @@ package client -// UserAdded is the event sent when a new user is added to a company +// UserAdded is the event sent when a new user is added to a company. +// SequenceNo is authz-service's global event sequence number; it orders the events. type UserAdded struct { - Email string `json:"email"` - CompanyID string `json:"companyId"` + Email string `json:"email"` + CompanyID string `json:"companyId"` + SequenceNo int `json:"sequenceNo"` } // UserRemoved is the event sent when a user is removed from a company type UserRemoved struct { - Email string `json:"email"` - CompanyID string `json:"companyId"` + Email string `json:"email"` + CompanyID string `json:"companyId"` + SequenceNo int `json:"sequenceNo"` } // Privilege is an enumeration of all available privileges @@ -51,14 +54,16 @@ func (e Privilege) String() string { // PrivilegeAdded is the event sent when a new privilege is added type PrivilegeAdded struct { - Email string `json:"email"` - CompanyID string `json:"companyId"` - Privilege Privilege `json:"privilege"` + Email string `json:"email"` + CompanyID string `json:"companyId"` + Privilege Privilege `json:"privilege"` + SequenceNo int `json:"sequenceNo"` } // PrivilegeRemoved is the event sent when a privilege is removed type PrivilegeRemoved struct { - Email string `json:"email"` - CompanyID string `json:"companyId"` - Privilege Privilege `json:"privilege"` + Email string `json:"email"` + CompanyID string `json:"companyId"` + Privilege Privilege `json:"privilege"` + SequenceNo int `json:"sequenceNo"` }