Press n or j to go to the next uncovered block, b, p or k for the previous block.
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 | 2x 2x 2x 2x 8x 8x 8x 3x 3x 2x 1x 2x | import {
Injectable,
Logger,
OnModuleDestroy,
OnModuleInit,
} from '@nestjs/common';
import { Kafka, Producer } from 'kafkajs';
import { TOPICS, WorkOrderEvent } from '@app/contracts';
/**
* Appends every domain event to the Kafka audit stream, keyed by the
* aggregate id: all events of one work order share a partition and so keep
* their order. Called from the outbox relay, never the write path — a failure
* here throws so the relay leaves the row unpublished and retries.
*/
@Injectable()
export class WorkOrderAuditProducer implements OnModuleInit, OnModuleDestroy {
private readonly logger = new Logger(WorkOrderAuditProducer.name);
private readonly producer: Producer;
constructor() {
const kafka = new Kafka({
clientId: 'work-orders-service',
brokers: (process.env.KAFKA_BROKERS ?? 'localhost:9092').split(','),
});
this.producer = kafka.producer({ allowAutoTopicCreation: true });
}
async onModuleInit(): Promise<void> {
try {
await this.producer.connect();
} catch (error) {
this.logger.error(
'kafka connect failed, audit records will be dropped',
error instanceof Error ? error.stack : String(error),
);
}
}
async onModuleDestroy(): Promise<void> {
await this.producer.disconnect();
}
async record(event: WorkOrderEvent): Promise<void> {
await this.producer.send({
topic: TOPICS.WORK_ORDER_EVENTS,
messages: [{ key: event.data.workOrderId, value: JSON.stringify(event) }],
});
}
}
|