request masuk -> validasi credential/token -> SecurityContext -> handler

Who produced this event?
Who was the original business actor?
Is the event allowed to represent that actor?
Is this event fresh, replayed, duplicated, or stale?
Can the consumer trust the identity metadata?

Consumer menerima event dengan actorId=user-123.
Consumer menganggap event itu benar karena ada actorId.

Broker authentication proves producer/client identity.
Event metadata states business actor identity.
Consumer policy decides whether producer is allowed to assert that actor identity.

{
  "specversion": "1.0",
  "id": "evt_01HZP9VKFZ5M8S6B2V0J6C4P8H",
  "source": "urn:service:case-api",
  "type": "reg.case.created.v1",
  "time": "2026-07-03T10:15:30Z",
  "subject": "case/case_123",
  "tenantid": "tenant_a",
  "actor_type": "USER",
  "actor_id": "user_123",
  "actor_session_id": "sess_789",
  "auth_time": "2026-07-03T10:10:12Z",
  "auth_assurance": "aal2",
  "auth_methods": ["password", "totp"],
  "producer_client_id": "case-api",
  "correlation_id": "corr_abc",
  "causation_id": "cmd_xyz",
  "datacontenttype": "application/json",
  "data": {
    "caseId": "case_123",
    "createdBy": "user_123"
  }
}

{
  "eventType": "CaseApproved",
  "accessToken": "eyJhbGciOi...",
  "caseId": "case_123"
}

Event may carry an actor snapshot.
Event must not carry bearer credentials.

{
  "actor_type": "USER",
  "actor_id": "user_123",
  "auth_assurance": "aal2",
  "auth_time": "2026-07-03T10:10:12Z"
}

access token
refresh token
session id raw
password
API key
TOTP secret
recovery code
private key
full cookie header
Authorization header

At time T, actor A caused business event E under authentication context C.

Actor A is still authenticated now.

Event identity is evidence of causation.
It is not a universal authorization grant.

public enum ActorType {
    USER,
    SERVICE,
    SYSTEM,
    JOB
}

public record AuthenticatedActor(
        ActorType actorType,
        String actorId,
        String tenantId,
        String sessionId,
        Instant authTime,
        String assuranceLevel,
        List<String> authenticationMethods,
        String clientId
) {
    public AuthenticatedActor {
        Objects.requireNonNull(actorType, "actorType");
        Objects.requireNonNull(actorId, "actorId");
        Objects.requireNonNull(tenantId, "tenantId");
        Objects.requireNonNull(authTime, "authTime");
        authenticationMethods = List.copyOf(authenticationMethods == null ? List.of() : authenticationMethods);
    }
}

public record EventIdentity(
        String eventId,
        String source,
        String type,
        String subject,
        String tenantId,
        AuthenticatedActor actor,
        String correlationId,
        String causationId,
        Instant occurredAt
) {}

public final class AuthenticatedActorFactory {

    public AuthenticatedActor from(Authentication authentication, String tenantId) {
        if (authentication == null || !authentication.isAuthenticated()) {
            throw new IllegalStateException("Authenticated request required");
        }

        if (authentication instanceof JwtAuthenticationToken jwtAuth) {
            Jwt jwt = jwtAuth.getToken();
            return new AuthenticatedActor(
                    ActorType.USER,
                    jwt.getSubject(),
                    tenantId,
                    claim(jwt, "sid"),
                    jwt.getIssuedAt() != null ? jwt.getIssuedAt() : Instant.now(),
                    claim(jwt, "acr"),
                    listClaim(jwt, "amr"),
                    claim(jwt, "azp")
            );
        }

        return new AuthenticatedActor(
                ActorType.USER,
                authentication.getName(),
                tenantId,
                null,
                Instant.now(),
                "unknown",
                List.of(authentication.getClass().getSimpleName()),
                null
        );
    }

    private static String claim(Jwt jwt, String name) {
        Object value = jwt.getClaims().get(name);
        return value == null ? null : value.toString();
    }

    @SuppressWarnings("unchecked")
    private static List<String> listClaim(Jwt jwt, String name) {
        Object value = jwt.getClaims().get(name);
        if (value instanceof Collection<?> values) {
            return values.stream().map(Object::toString).toList();
        }
        return List.of();
    }
}

Authentication context is captured once at command boundary.
Outbox relay only publishes already captured event metadata.
Relay must not invent actor identity from its own service account.

create table event_outbox (
    outbox_id uuid primary key,
    aggregate_type text not null,
    aggregate_id text not null,
    event_type text not null,
    event_version int not null,
    tenant_id text not null,

    actor_type text not null,
    actor_id text not null,
    actor_session_id text null,
    actor_auth_time timestamptz null,
    actor_assurance text null,
    actor_methods jsonb not null default '[]',
    actor_client_id text null,

    correlation_id text not null,
    causation_id text null,
    payload jsonb not null,
    headers jsonb not null default '{}',

    created_at timestamptz not null default now(),
    published_at timestamptz null,
    publish_attempts int not null default 0
);

create index idx_event_outbox_pending
    on event_outbox(created_at)
    where published_at is null;

HTTP request thread punya SecurityContext.
Outbox relay thread tidak punya SecurityContext user.

@Transactional
public CaseId createCase(CreateCaseCommand command, AuthenticatedActor actor) {
    CaseEntity saved = caseRepository.save(CaseEntity.newDraft(
            command.caseType(),
            actor.tenantId(),
            actor.actorId()
    ));

    EventIdentity identity = new EventIdentity(
            EventIds.newId(),
            "urn:service:case-api",
            "reg.case.created.v1",
            "case/" + saved.id(),
            actor.tenantId(),
            actor,
            command.correlationId(),
            command.commandId(),
            Instant.now()
    );

    outboxRepository.append(identity, Map.of(
            "caseId", saved.id().toString(),
            "caseType", saved.caseType()
    ));

    return saved.id();
}

Map<String, Object> props = new HashMap<>();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-1:9093,kafka-2:9093");
props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
props.put(SaslConfigs.SASL_MECHANISM, "SCRAM-SHA-512");
props.put(SaslConfigs.SASL_JAAS_CONFIG, """
    org.apache.kafka.common.security.scram.ScramLoginModule required
    username="case-api"
    password="${KAFKA_CASE_API_SECRET}";
    """);
props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, "/etc/security/kafka/truststore.p12");
props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, System.getenv("KAFKA_TRUSTSTORE_PASSWORD"));

Do not hard-code JAAS secrets in source code.
Prefer secret manager / mounted secret / callback handler.
Rotate producer credentials independently per service.

case-api may write reg.case.* topics.
case-api must not write reg.payment.* topics.
notification-service may read case events.
notification-service must not write case decision events.

1. Validate broker client context if available.
2. Validate topic -> event type mapping.
3. Validate envelope fields.
4. Validate schema version.
5. Validate tenant boundary.
6. Validate producer trust policy.
7. Check idempotency/replay.
8. Execute business handler.

public final class EventEnvelopeValidator {

    private final ProducerTrustPolicy producerTrustPolicy;
    private final ProcessedEventRepository processedEvents;

    public void validate(EventEnvelope event, ConsumerContext consumerContext) {
        requireNonBlank(event.id(), "event.id");
        requireNonBlank(event.source(), "event.source");
        requireNonBlank(event.type(), "event.type");
        requireNonBlank(event.tenantId(), "event.tenantId");
        requireNonBlank(event.actor().actorId(), "actor.actorId");
        requireNonBlank(event.correlationId(), "correlationId");

        if (!producerTrustPolicy.canAssert(
                event.source(),
                event.type(),
                event.tenantId(),
                event.actor().actorType())) {
            throw new RejectedEventException("Producer is not trusted for event identity");
        }

        if (processedEvents.alreadyProcessed(event.id())) {
            throw new DuplicateEventException(event.id());
        }
    }

    private static void requireNonBlank(String value, String field) {
        if (value == null || value.isBlank()) {
            throw new RejectedEventException("Missing " + field);
        }
    }
}

trustedEventSources:
  - source: "urn:service:case-api"
    eventTypes:
      - "reg.case.created.v1"
      - "reg.case.submitted.v1"
    actorTypes:
      - "USER"
      - "SERVICE"
    tenants: "*"

  - source: "urn:service:scheduler"
    eventTypes:
      - "reg.case.sla.expired.v1"
    actorTypes:
      - "SYSTEM"
    tenants: "*"

Not every producer may assert every actor type.

{
  "type": "reg.case.autoClosed.v1",
  "actor_id": "admin-user-1"
}

{
  "type": "reg.case.autoClosed.v1",
  "actor_type": "SYSTEM",
  "actor_id": "sla-auto-close-policy",
  "causation_id": "timer_123",
  "policy_version": "sla-close-v7"
}

{
  "tenantid": "tenant_a",
  "actor_id": "user_123",
  "subject": "case/case_123"
}

event.tenantId must match aggregate tenant boundary.

if (!event.tenantId().equals(caseEntity.tenantId())) {
    throw new RejectedEventException("Tenant mismatch: event cannot mutate aggregate");
}

create table processed_event (
    consumer_name text not null,
    event_id text not null,
    tenant_id text not null,
    event_type text not null,
    source text not null,
    actor_id text not null,
    processed_at timestamptz not null default now(),
    primary key (consumer_name, event_id)
);

@Transactional
public void handle(EventEnvelope event) {
    validator.validate(event, ConsumerContext.current());

    if (!processedEventRepository.tryMarkProcessing("notification-service", event)) {
        return; // duplicate; safe no-op
    }

    notificationService.sendCaseCreatedNotification(event);
}

signature = HMAC(secret, canonical(event headers + payload hash))

signature = Sign(privateKey, canonical(event headers + payload hash))
consumer verifies with producer public key

public final class EventHeaders {
    public static final String EVENT_ID = "ce_id";
    public static final String EVENT_SOURCE = "ce_source";
    public static final String EVENT_TYPE = "ce_type";
    public static final String TENANT_ID = "ce_tenantid";
    public static final String ACTOR_TYPE = "ce_actor_type";
    public static final String ACTOR_ID = "ce_actor_id";
    public static final String CORRELATION_ID = "correlation_id";

    private EventHeaders() {}
}

public ProducerRecord<String, byte[]> toRecord(EventEnvelope event, byte[] payload) {
    ProducerRecord<String, byte[]> record = new ProducerRecord<>(
            topicFor(event.type()),
            event.tenantId() + ":" + event.subject(),
            payload
    );

    add(record, EventHeaders.EVENT_ID, event.id());
    add(record, EventHeaders.EVENT_SOURCE, event.source());
    add(record, EventHeaders.EVENT_TYPE, event.type());
    add(record, EventHeaders.TENANT_ID, event.tenantId());
    add(record, EventHeaders.ACTOR_TYPE, event.actor().actorType().name());
    add(record, EventHeaders.ACTOR_ID, event.actor().actorId());
    add(record, EventHeaders.CORRELATION_ID, event.correlationId());

    return record;
}

@KafkaListener(topics = "reg.case-events")
public void onMessage(ConsumerRecord<String, byte[]> record) {
    EventEnvelope envelope = envelopeReader.from(record.headers(), record.value());
    eventEnvelopeValidator.validate(envelope, ConsumerContext.kafka(record));
    caseEventHandler.handle(envelope, record.value());
}

public final class EventActorAuthentication extends AbstractAuthenticationToken {
    private final AuthenticatedActor actor;

    public EventActorAuthentication(AuthenticatedActor actor, Collection<GrantedAuthority> authorities) {
        super(authorities);
        this.actor = actor;
        setAuthenticated(true);
    }

    @Override
    public Object getCredentials() {
        return "event-actor-snapshot";
    }

    @Override
    public Object getPrincipal() {
        return actor;
    }

    @Override
    public String getName() {
        return actor.actorId();
    }
}

public void withEventSecurityContext(EventEnvelope event, Runnable handler) {
    SecurityContext previous = SecurityContextHolder.getContext();
    SecurityContext context = SecurityContextHolder.createEmptyContext();

    try {
        context.setAuthentication(new EventActorAuthentication(event.actor(), List.of()));
        SecurityContextHolder.setContext(context);
        handler.run();
    } finally {
        SecurityContextHolder.clearContext();
        SecurityContextHolder.setContext(previous);
    }
}

An event actor snapshot is not equivalent to a freshly authenticated user.
Do not grant live interactive privileges from it.

Is this consumer allowed to process this event and mutate this aggregate?

public void apply(CaseApproved event) {
    CaseRecord caseRecord = caseRepository.require(event.caseId());

    if (!caseRecord.tenantId().equals(event.tenantId())) {
        throw new RejectedEventException("Cross-tenant event rejected");
    }

    if (!caseRecord.isSubmitted()) {
        throw new RejectedEventException("Cannot approve non-submitted case");
    }

    // proceed
}

Day 1: user submits enforcement case.
Day 3: risk scoring completes.
Day 7: decision workflow auto-escalates.
Day 10: user account is disabled.

The user caused submission.
The risk service caused scoring.
The policy caused escalation.

Replay actor = operator who initiated replay.
Original actor = actor in original event.

{
  "original_actor_id": "user_123",
  "replay_actor_id": "ops_456",
  "replay_reason": "fixed schema mapping bug",
  "replay_time": "2026-07-03T12:00:00Z"
}

events.accepted.count{type,source,tenant}
events.rejected.count{reason,type,source}
events.duplicate.count{consumer,type}
events.cross_tenant_rejected.count{source,type}
events.missing_actor.count{source,type}
events.dlq.count{reason,type}

source not in allowlist starts producing events
sudden spike in cross-tenant rejection
event type appears on unexpected topic
missing actor metadata after deployment
DLQ contains security-sensitive event type

missing actor -> reject
missing tenant -> reject
unknown source -> reject
source cannot assert USER actor -> reject
duplicate event id -> no-op
cross-tenant aggregate mutation -> reject

event type
schema version
required identity metadata
tenant metadata
auth assurance representation
correlation/causation id

{
  "id": "evt_attack_1",
  "source": "urn:service:notification-service",
  "type": "reg.case.approved.v1",
  "tenantid": "tenant_b",
  "actor_type": "USER",
  "actor_id": "admin-user"
}

Rejected because notification-service cannot assert case.approved.
Rejected because aggregate tenant mismatch.

same event twice -> side effect once
same event after consumer restart -> side effect once
DLQ replay -> original actor preserved + replay operator recorded

[ ] Producer identity is authenticated at broker or gateway boundary.
[ ] Producer is authorized only for required topics.
[ ] Event envelope includes event id, source, type, time, tenant, actor, correlation id.
[ ] Bearer tokens, sessions, cookies, API keys, and secrets are never placed in events.
[ ] Consumer validates producer trust policy before business handling.
[ ] Consumer validates tenant boundary before aggregate mutation.
[ ] Event idempotency is implemented per consumer.
[ ] DLQ replay preserves original actor and records replay operator.
[ ] Long-running workflows model USER, SERVICE, SYSTEM, and JOB actors explicitly.
[ ] Audit logs include event id, source, actor, tenant, consumer, topic, partition, offset.
[ ] Security tests include spoofed actor, wrong source, wrong tenant, duplicate event.
[ ] Observability alerts detect unknown source, missing metadata, cross-tenant rejection.

Broker authenticates technical producer.
Envelope carries business actor snapshot.
Consumer validates source, tenant, event type, and producer trust.
Business handler enforces domain invariants.
Audit records causation without pretending the original session still exists.

producer identity != actor identity
actor snapshot != live login
event source != tenant
replay operator != original actor
service action != user action