IAM Service -- Event Catalog

Overview

The IAM service publishes domain events whenever a state change occurs. Events represent facts (something that happened), not intentions. They are published via a transactional outbox: the domain write and the outbox insert happen in a single database transaction, guaranteeing that every committed state change produces exactly one event record.

A background relay publishes pending outbox records to RabbitMQ. Delivery is at-least-once -- the same event may be delivered more than once. Consumers must be idempotent.

Events are only emitted on real state changes. Idempotent no-op retries (e.g., suspending an already-suspended tenant) do not produce events.

---

Connection Setup

Exchange

Property	Value
Exchange name	iam.events
Type	topic
Durable	true
Auto-delete	false

Queue Declaration

Consumers must declare their own durable queue and bind it to the exchange. Queue names should be unique per consuming service (e.g., billing-iam-consumer, notifications-iam-consumer).

Binding Patterns

The routing key format is {aggregate_type}.{event_type}.

Pattern	Matches
#	All events
tenant.*	tenant.created, tenant.suspended, tenant.reactivated
tenant.created	Only tenant creation events
membership.*	membership.created, membership.suspended, membership.reactivated
invitation.*	invitation.accepted, invitation.revoked
user.*	user.created, user.suspended, user.reactivated
membership.user.role.*	membership.user.role.assigned, membership.user.role.unassigned
invitation.user.invited	invitation.user.invited
realm.*	realm.created
role.*	role.created
permission.*	permission.created

Note: The routing key is constructed as {aggregate_type}.{event_type}. For events where the event type itself contains dots (e.g., user.role.assigned on the membership aggregate), the full routing key becomes membership.user.role.assigned. Plan your binding patterns accordingly.

---

Message Structure

Each AMQP message is structured as follows:

AMQP Properties

Property	Description	Example
MessageId	Unique event UUID. Use for deduplication.	f47ac10b-58cc-4372-a567-0e02b2c3d479
ContentType	Always application/json.	application/json
Timestamp	When the event occurred.	2025-01-15T10:30:00Z

AMQP Headers

Header	Type	Always Present	Description
event_type	string	Yes	The event type (e.g., tenant.created).
aggregate_type	string	Yes	The aggregate that owns this event (e.g., tenant).
aggregate_id	string	Yes	UUID of the aggregate instance.
tenant_id	string	No	UUID of the tenant scope. Absent for global resources (users, realms, permissions).
occurred_at	string	Yes	RFC3339Nano timestamp.
schema_version	int32	Yes	Payload schema version. Currently 1.

Body

The body is a JSON object containing domain-specific fields. The exact fields depend on the event type and are documented in the catalog below.

Example Raw Message

Properties:
  MessageId:   "f47ac10b-58cc-4372-a567-0e02b2c3d479"
  ContentType: "application/json"
  Timestamp:   2025-01-15T10:30:00Z

Headers:
  event_type:     "tenant.created"
  aggregate_type: "tenant"
  aggregate_id:   "a1b2c3d4-e5f6-7890-abcd-ef1234567890"
  tenant_id:      "a1b2c3d4-e5f6-7890-abcd-ef1234567890"
  occurred_at:    "2025-01-15T10:30:00.123456789Z"
  schema_version: 1

Body:
  {"tenant_id":"a1b2c3d4-...","realm_id":"...","slug":"acme","display_name":"Acme Corp"}

---

Event Catalog

Realm Events

realm.created

Property	Value
Event type	realm.created
Routing key	realm.realm.created
Aggregate	realm
Tenant scoped	No
Fires when	A new realm is created.

Payload fields:

Field	Type	Description
ID	string	UUID of the realm.
Key	string	Unique key for the realm.
Name	string	Display name of the realm.
CreatedAt	string	RFC3339 timestamp of creation.

The payload is the JSON-serialized Realm struct. Field names use Go default casing (no JSON tags).

---

Tenant Events

tenant.created

Property	Value
Event type	tenant.created
Routing key	tenant.tenant.created
Aggregate	tenant
Tenant scoped	Yes
Fires when	A new tenant is created.

Payload fields:

Field	Type	Description
tenant_id	string	UUID of the created tenant.
realm_id	string	UUID of the realm this tenant belongs to.
slug	string	URL-safe tenant identifier.
display_name	string	Human-readable tenant name.

---

tenant.suspended

Property	Value
Event type	tenant.suspended
Routing key	tenant.tenant.suspended
Aggregate	tenant
Tenant scoped	Yes
Fires when	An active tenant is suspended.

Payload fields:

Field	Type	Description
tenant_id	string	UUID of the suspended tenant.

---

tenant.reactivated

Property	Value
Event type	tenant.reactivated
Routing key	tenant.tenant.reactivated
Aggregate	tenant
Tenant scoped	Yes
Fires when	A suspended tenant is reactivated.

Payload fields:

Field	Type	Description
tenant_id	string	UUID of the reactivated tenant.

---

User Events

user.created

Property	Value
Event type	user.created
Routing key	user.user.created
Aggregate	user
Tenant scoped	No
Fires when	A new global user is created.

Payload fields:

Field	Type	Presence	Description
user_id	string	Always	UUID of the created user.
email	string	Optional	Email address, if provided.
phone_e164	string	Optional	Phone number in E.164 format, if provided.
display_name	string	Optional	Display name, if provided.

---

user.suspended

Property	Value
Event type	user.suspended
Routing key	user.user.suspended
Aggregate	user
Tenant scoped	No
Fires when	An active user is suspended.

Payload fields:

Field	Type	Description
user_id	string	UUID of the suspended user.

---

user.reactivated

Property	Value
Event type	user.reactivated
Routing key	user.user.reactivated
Aggregate	user
Tenant scoped	No
Fires when	A suspended user is reactivated.

Payload fields:

Field	Type	Description
user_id	string	UUID of the reactivated user.

---

Membership Events

membership.created

Property	Value
Event type	membership.created
Routing key	membership.membership.created
Aggregate	membership
Tenant scoped	Yes
Fires when	A user is added to a tenant (directly or via invitation acceptance).

Payload fields:

Field	Type	Description
membership_id	string	UUID of the created membership.
tenant_id	string	UUID of the tenant.
user_id	string	UUID of the user.

---

membership.suspended

Property	Value
Event type	membership.suspended
Routing key	membership.membership.suspended
Aggregate	membership
Tenant scoped	Yes
Fires when	An active membership is suspended.

Payload fields:

Field	Type	Description
membership_id	string	UUID of the suspended membership.

---

membership.reactivated

Property	Value
Event type	membership.reactivated
Routing key	membership.membership.reactivated
Aggregate	membership
Tenant scoped	Yes
Fires when	A suspended membership is reactivated.

Payload fields:

Field	Type	Description
membership_id	string	UUID of the reactivated membership.

---

Role and Permission Events

role.created

Property	Value
Event type	role.created
Routing key	role.role.created
Aggregate	role
Tenant scoped	Yes
Fires when	A new role is created within a tenant.

Payload fields:

Field	Type	Description
role_id	string	UUID of the created role.
tenant_id	string	UUID of the tenant that owns this role.
key	string	Unique key for the role within the tenant.
name	string	Human-readable role name.

---

permission.created

Property	Value
Event type	permission.created
Routing key	permission.permission.created
Aggregate	permission
Tenant scoped	No
Fires when	A new global permission is created.

Payload fields:

Field	Type	Presence	Description
ID	string	Always	UUID of the permission.
Key	string	Always	Unique permission key.
Description	string	Optional	Description of the permission (null if not set).
CreatedAt	string	Always	RFC3339 timestamp of creation.

The payload is the JSON-serialized Permission struct. Field names use Go default casing (no JSON tags).

---

user.role.assigned

Property	Value
Event type	user.role.assigned
Routing key	membership.user.role.assigned
Aggregate	membership
Tenant scoped	Yes
Fires when	A role is assigned to a membership.

Payload fields:

Field	Type	Description
assignment_id	string	UUID of the role assignment record.
membership_id	string	UUID of the membership receiving the role.
role_id	string	UUID of the assigned role.

---

user.role.unassigned

Property	Value
Event type	user.role.unassigned
Routing key	membership.user.role.unassigned
Aggregate	membership
Tenant scoped	Yes
Fires when	A role is removed from a membership.

Payload fields:

Field	Type	Description
membership_id	string	UUID of the membership losing the role.
role_id	string	UUID of the unassigned role.

---

Invitation Events

user.invited

Property	Value
Event type	user.invited
Routing key	invitation.user.invited
Aggregate	invitation
Tenant scoped	Yes
Fires when	A new invitation is created for a user to join a tenant.

Payload fields:

Field	Type	Description
invitation_id	string	UUID of the invitation.
tenant_id	string	UUID of the tenant the user is invited to.
email	string	Email address of the invitee.
token	string	Raw invitation token (for email delivery).
expires_at	string	RFC3339 timestamp of when the invitation expires.

Security note: The token field contains the raw invitation token. This event should only be consumed by the notification/email service. Do not log or persist this value beyond its intended use.

---

invitation.accepted

Property	Value
Event type	invitation.accepted
Routing key	invitation.invitation.accepted
Aggregate	invitation
Tenant scoped	Yes
Fires when	An invitation is accepted by a user. A membership.created event is also emitted in the same transaction.

Payload fields:

Field	Type	Description
invitation_id	string	UUID of the accepted invitation.
user_id	string	UUID of the user who accepted.

---

invitation.revoked

Property	Value
Event type	invitation.revoked
Routing key	invitation.invitation.revoked
Aggregate	invitation
Tenant scoped	Yes
Fires when	A pending invitation is revoked.

Payload fields:

Field	Type	Description
invitation_id	string	UUID of the revoked invitation.

---

API Key Events

api_key.created

Property	Value
Event type	api_key.created
Routing key	api_key.api_key.created
Aggregate	api_key
Tenant scoped	No
Fires when	A new database-backed API key is created.

Payload fields:

Field	Type	Description
api_key_id	string	UUID of the created API key.
name	string	Human-readable name of the key.
key_prefix	string	First 8 characters of the raw key (for identification).

Note: The raw key is never included in events.

---

api_key.revoked

Property	Value
Event type	api_key.revoked
Routing key	api_key.api_key.revoked
Aggregate	api_key
Tenant scoped	No
Fires when	An active API key is revoked. Not emitted if the key was already revoked.

Payload fields:

Field	Type	Description
api_key_id	string	UUID of the revoked API key.
name	string	Human-readable name of the key.

---

Event Summary Table

#	Event Type	Routing Key	Aggregate	Tenant Scoped
1	realm.created	realm.realm.created	realm	No
2	tenant.created	tenant.tenant.created	tenant	Yes
3	tenant.suspended	tenant.tenant.suspended	tenant	Yes
4	tenant.reactivated	tenant.tenant.reactivated	tenant	Yes
5	user.created	user.user.created	user	No
6	user.suspended	user.user.suspended	user	No
7	user.reactivated	user.user.reactivated	user	No
8	membership.created	membership.membership.created	membership	Yes
9	membership.suspended	membership.membership.suspended	membership	Yes
10	membership.reactivated	membership.membership.reactivated	membership	Yes
11	role.created	role.role.created	role	Yes
12	permission.created	permission.permission.created	permission	No
13	user.role.assigned	membership.user.role.assigned	membership	Yes
14	user.role.unassigned	membership.user.role.unassigned	membership	Yes
15	user.invited	invitation.user.invited	invitation	Yes
16	invitation.accepted	invitation.invitation.accepted	invitation	Yes
17	invitation.revoked	invitation.invitation.revoked	invitation	Yes
18	api_key.created	api_key.api_key.created	api_key	No
19	api_key.revoked	api_key.api_key.revoked	api_key	No

---

Consumer Example

The following Go program connects to RabbitMQ, declares a durable queue, binds to all IAM events, and processes them with manual acknowledgement and deduplication.

package main

import (
	"encoding/json"
	"log"
	"os"
	"os/signal"
	"sync"
	"syscall"

	amqp "github.com/rabbitmq/amqp091-go"
)

func headerString(headers amqp.Table, key string) string {
	v, ok := headers[key]
	if !ok {
		return ""
	}
	s, _ := v.(string)
	return s
}

func main() {
	rabbitURL := os.Getenv("RABBITMQ_URL")
	if rabbitURL == "" {
		rabbitURL = "amqp://guest:guest@localhost:5672/"
	}

	conn, err := amqp.Dial(rabbitURL)
	if err != nil {
		log.Fatalf("Failed to connect to RabbitMQ: %v", err)
	}
	defer conn.Close()

	ch, err := conn.Channel()
	if err != nil {
		log.Fatalf("Failed to open channel: %v", err)
	}
	defer ch.Close()

	// Declare the exchange (idempotent -- safe to call even if it already exists)
	err = ch.ExchangeDeclare(
		"iam.events", // name
		"topic",      // type
		true,         // durable
		false,        // auto-delete
		false,        // internal
		false,        // no-wait
		nil,          // arguments
	)
	if err != nil {
		log.Fatalf("Failed to declare exchange: %v", err)
	}

	// Declare a durable queue for this consumer
	q, err := ch.QueueDeclare(
		"my-service-iam-consumer", // name -- unique per consuming service
		true,                      // durable
		false,                     // auto-delete
		false,                     // exclusive
		false,                     // no-wait
		nil,                       // arguments
	)
	if err != nil {
		log.Fatalf("Failed to declare queue: %v", err)
	}

	// Bind to all IAM events. Use a more specific pattern to filter:
	//   "tenant.*"                    -- all tenant events
	//   "membership.membership.*"     -- membership lifecycle events
	//   "membership.user.role.*"      -- role assignment events
	err = ch.QueueBind(
		q.Name,       // queue name
		"#",          // routing key pattern
		"iam.events", // exchange
		false,        // no-wait
		nil,          // arguments
	)
	if err != nil {
		log.Fatalf("Failed to bind queue: %v", err)
	}

	// Use manual acknowledgement (autoAck: false)
	msgs, err := ch.Consume(
		q.Name, // queue
		"",     // consumer tag (auto-generated)
		false,  // autoAck -- set to false for manual ack
		false,  // exclusive
		false,  // noLocal
		false,  // noWait
		nil,    // arguments
	)
	if err != nil {
		log.Fatalf("Failed to register consumer: %v", err)
	}

	log.Printf("Consumer started. Listening on queue '%s'", q.Name)

	// Graceful shutdown
	sigChan := make(chan os.Signal, 1)
	signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)

	var wg sync.WaitGroup
	wg.Add(1)

	go func() {
		defer wg.Done()
		for msg := range msgs {
			eventType := headerString(msg.Headers, "event_type")
			aggregateID := headerString(msg.Headers, "aggregate_id")
			tenantID := headerString(msg.Headers, "tenant_id")

			log.Printf("Received event: type=%s aggregate_id=%s tenant_id=%s message_id=%s",
				eventType, aggregateID, tenantID, msg.MessageId)

			// --- Deduplication ---
			// Use msg.MessageId to check if this event was already processed.
			// Store processed MessageIds in your database within the same
			// transaction as your side effects.

			// --- Handle by event type ---
			switch eventType {
			case "tenant.created":
				var payload struct {
					TenantID    string `json:"tenant_id"`
					RealmID     string `json:"realm_id"`
					Slug        string `json:"slug"`
					DisplayName string `json:"display_name"`
				}
				if err := json.Unmarshal(msg.Body, &payload); err != nil {
					log.Printf("Failed to unmarshal payload: %v", err)
					_ = msg.Nack(false, false) // discard malformed messages
					continue
				}
				log.Printf("Tenant created: %s (%s)", payload.DisplayName, payload.Slug)

			// Handle other event types ...

			default:
				log.Printf("Unhandled event type: %s", eventType)
			}

			// Acknowledge after successful processing
			if err := msg.Ack(false); err != nil {
				log.Printf("Failed to ack message: %v", err)
			}
		}
	}()

	<-sigChan
	log.Println("Shutting down consumer...")

	// Cancel the consumer so the msgs channel closes
	_ = ch.Cancel("", false)
	wg.Wait()

	log.Println("Consumer stopped.")
}

---

Consumer Best Practices

Idempotency

Delivery is at-least-once. The same event will be delivered more than once during retries, rebalances, or publisher relay restarts.

• Use the MessageId AMQP property (a UUID) to deduplicate.
• Store processed MessageId values in your database within the same transaction as your side effects.
• A simple approach: maintain a processed_events table with a unique constraint on message_id, and INSERT ... ON CONFLICT DO NOTHING before processing.

CREATE TABLE processed_events (
    message_id UUID PRIMARY KEY,
    processed_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

Error Handling

• Transient errors (database timeouts, network blips): Nack with requeue: true. The message will be redelivered.
• Permanent errors (malformed payload, unknown event type from an older schema): Nack with requeue: false. The message goes to a dead-letter queue if configured, or is discarded.
• Never silently drop messages. Log every Nack with the MessageId and the reason.

Acknowledgement Strategy

• Use manual acknowledgement (autoAck: false).
• Ack only after your side effects are committed.
• If your processing crashes before Ack, the message is automatically redelivered -- this is the intended behavior.
• Avoid long-running processing in the message handler. If you need to do heavy work, persist the event and process it asynchronously.

Graceful Shutdown

• Catch SIGINT and SIGTERM.
• Cancel the AMQP consumer so the delivery channel closes.
• Wait for in-flight message processing to complete before closing the connection.
• Unacknowledged messages will be redelivered to another consumer (or the same one on restart).

Schema Evolution

• The schema_version header indicates the payload schema version (currently 1).
• When consuming events, check the version and handle unknown versions gracefully (log and skip, or Nack without requeue).
• New fields may be added to payloads in a backward-compatible way. Use lenient JSON deserialization and do not fail on unknown fields.

Ordering

• RabbitMQ does not guarantee strict global ordering across multiple publishers or queues.
• Events for the same aggregate are published sequentially by the outbox relay, but network conditions may cause reordering.
• Do not rely on event ordering for correctness. Use the occurred_at header and aggregate version checks if ordering matters for your use case.