Copilot instructions

1---
2applyTo: "**/events/**/*.java, **/messaging/**/*.java, **/kafka/**/*.java, **/listeners/**/*.java, **/publishers/**/*.java, **/events/**/*.ts, **/messaging/**/*.ts"
3---
4 
5## Context
6 
7This instruction file applies to event-driven and messaging code: Kafka producers, Kafka consumers, Spring `@EventListener` / `ApplicationEventPublisher`, AWS SNS/SQS integrations, and Outbox Pattern implementations. Events are the primary integration mechanism between bounded contexts. All events must have a versioned schema registered with the Schema Registry. Consumers must be idempotent and route failures to a Dead-Letter Queue.
8 
9---
10 
11## Coding Standards
12 
13- **Event envelope:** Every domain event carries `eventId` (UUID), `eventType`, `aggregateId`, `aggregateType`, `occurredAt` (ISO 8601 UTC), `schemaVersion`, and `payload`
14- **Naming convention:** `{domain}.{entity}.{verb}` in past tense — e.g. `orders.order.placed`, `payments.payment.authorised`
15- **Schema evolution:** Only additive changes (add nullable fields); never remove or rename fields without a deprecation lifecycle
16- **Outbox Pattern:** Publish events via a transactional outbox table — never produce directly inside a `@Transactional` method
17- **Idempotent consumer:** Track processed `eventId` values in a deduplication store; skip already-processed events
18- **DLQ routing:** After 3 retry attempts, route to `{topic}.dlq` with error metadata headers
19- **Manual offset commit:** `enable.auto.commit=false`; commit only after successful processing and DLQ routing
20- **No event sourcing without explicit approval:** CQRS/ES adds significant complexity; get ARB sign-off before introducing
21 
22---
23 
24## Preferred Patterns
25 
26### Event Envelope (Java)
27 
28```java
29// ✅ CORRECT — structured event envelope
30public record DomainEvent<T>(
31    UUID eventId,
32    String eventType,
33    String aggregateId,
34    String aggregateType,
35    Instant occurredAt,
36    int schemaVersion,
37    T payload
38) {
39    public static <T> DomainEvent<T> of(String eventType, String aggregateId,
40                                         String aggregateType, T payload) {
41        return new DomainEvent<>(
42            UUID.randomUUID(), eventType, aggregateId, aggregateType,
43            Instant.now(), 1, payload
44        );
45    }
46}
47 
48// Usage
49var event = DomainEvent.of("orders.order.placed", order.getId().toString(), "Order", new OrderPlacedPayload(order));
50```
51 
52### Outbox Pattern
53 
54```java
55// ✅ CORRECT — write event to outbox table in same transaction as aggregate
56@Service
57@RequiredArgsConstructor
58public class OrderService {
59 
60    private final OrderRepository orderRepo;
61    private final OutboxRepository outboxRepo;
62 
63    @Transactional
64    public Order placeOrder(PlaceOrderCommand cmd) {
65        Order order = Order.create(cmd);
66        orderRepo.save(order);
67 
68        OutboxEntry entry = OutboxEntry.from(
69            DomainEvent.of("orders.order.placed", order.getId().toString(), "Order",
70                new OrderPlacedPayload(order))
71        );
72        outboxRepo.save(entry);  // same transaction — atomically consistent
73 
74        return order;
75    }
76}
77 
78// ❌ WRONG — publishing directly in @Transactional risks dual-write inconsistency
79@Transactional
80public Order placeOrder(PlaceOrderCommand cmd) {
81    Order order = orderRepo.save(Order.create(cmd));
82    kafkaTemplate.send("orders.placed", order.getId().toString(), payload); // may succeed after DB rollback
83    return order;
84}
85```
86 
87### Idempotent Consumer
88 
89```java
90// ✅ CORRECT — deduplication via processed event store
91@KafkaListener(topics = "orders.order.placed", groupId = "inventory-service")
92public void onOrderPlaced(ConsumerRecord<String, OrderPlacedEvent> record) {
93    String eventId = record.headers().lastHeader("eventId").toString();
94 
95    if (processedEventStore.exists(eventId)) {
96        log.info("Skipping duplicate event: eventId={}", eventId);
97        return;
98    }
99 
100    try {
101        inventoryService.reserveStock(record.value());
102        processedEventStore.markProcessed(eventId);
103    } catch (RetryableException exc) {
104        throw exc;  // Spring Kafka retries
105    } catch (Exception exc) {
106        log.error("Non-retryable failure; routing to DLQ: eventId={}", eventId, exc);
107        dlqProducer.send(record.topic() + ".dlq", record.key(), record.value(), _errorHeaders(eventId, exc));
108    }
109}
110```
111 
112### Retry + DLQ Configuration (Spring Kafka)
113 
114```java
115// ✅ CORRECT — fixed retry with DLQ after exhaustion
116@Bean
117public DefaultErrorHandler errorHandler(KafkaTemplate<String, ?> template) {
118    var dlqPublisher = new DeadLetterPublishingRecoverer(template,
119        (rec, ex) -> new TopicPartition(rec.topic() + ".dlq", rec.partition()));
120 
121    var backOff = new FixedBackOff(2_000L, 3L);  // 2s delay, 3 attempts
122    return new DefaultErrorHandler(dlqPublisher, backOff);
123}
124```
125 
126---
127 
128## Anti-Patterns — Do NOT Generate
129 
130```java
131// WRONG: direct Kafka produce inside @Transactional [BLOCKER]
132@Transactional
133public void placeOrder(PlaceOrderCommand cmd) {
134    orderRepo.save(order);
135    kafkaTemplate.send("orders.placed", payload);  // dual-write — can desync
136}
137 
138// WRONG: no deduplication — consumer is not idempotent [BLOCKER]
139@KafkaListener(topics = "orders.order.placed")
140public void handle(OrderPlacedEvent event) {
141    inventoryService.reserveStock(event);  // runs twice if message replayed
142}
143 
144// WRONG: silently swallowing failures [BLOCKER]
145try {
146    process(record);
147} catch (Exception e) {
148    log.warn("Failed to process");  // no DLQ, record lost
149}
150 
151// WRONG: past-tense violation — event named in present tense [MINOR]
152"orders.order.place"  // should be "orders.order.placed"
153 
154// WRONG: breaking schema change — removing a field [MAJOR]
155// Removing 'currency' field from OrderPlacedPayload — breaks existing consumers
156```
157 
158---
159 
160## Dependencies & Versions
161 
162| Technology | Version | Notes |
163|-----------|---------|-------|
164| spring-kafka | 3.x | `@KafkaListener`, `DefaultErrorHandler`, `DeadLetterPublishingRecoverer` |
165| confluent-kafka-java | 7.x | `KafkaProducer` / `KafkaConsumer` with Schema Registry |
166| avro | 1.11+ | Schema Registry Avro serialisation; use specific record types |
167| resilience4j | 2.x | Circuit breaker for downstream calls triggered by events |
168| aws-java-sdk-sqs | 2.x | `SqsAsyncClient`; visibility timeout ≥ max processing time |
169| aws-java-sdk-sns | 2.x | Fan-out pattern; `MessageAttributes` for filtering |
170 
171---
172 
173## Test Conventions
174 
175- Integration test consumers with `@EmbeddedKafka` or Testcontainers Kafka container
176- Test idempotency: publish the same event twice and assert the downstream effect occurs once
177- Test DLQ routing: configure retries to 0 and inject a failing handler; assert message appears in `.dlq` topic
178- Test Outbox relay: commit a transaction containing an outbox entry; assert the relay process publishes to Kafka
179- Verify event envelope fields: `eventId`, `occurredAt`, `schemaVersion` are populated on every published event
180- Use `@MockBean` to isolate consumer logic from downstream service calls in unit tests
181