change kafka broker settings

apps/service-auth/src/app.module.ts

@@ -2,10 +2,11 @@ import { KafkaConsumerModule, KafkaProducerModule } from '@app/kafka';
 import { MongoDbModule } from '@app/mongodb';
 import { RedisModule } from '@app/redis';
 import { Module } from '@nestjs/common';
-import { ConfigModule } from '@nestjs/config';
+import { ConfigModule, ConfigService } from '@nestjs/config';
 import { logLevel } from 'kafkajs';
 import { AuthModule } from './auth/auth.module';
 import { ClansModule } from './clans/clans.module';
+import { ConsumerModule } from './consumer/consumer.module';
 import { GuildsModule } from './guilds/guilds.module';
 import { LinksModule } from './links/links.module';
 import { PlayersModule } from './players/players.module';
@@ -21,31 +22,32 @@ import { PlayersModule } from './players/players.module';
  LinksModule,
  PlayersModule,
  KafkaProducerModule.forRootAsync({
- useFactory() {
+ useFactory: (configService: ConfigService) => {
  return {
  kafkaConfig: {
  clientId: 'kafka-client-id',
- brokers: ['localhost:9092'],
+ brokers: [configService.getOrThrow('KAFKA_BROKER')],
  logLevel: logLevel.NOTHING,
  },
  producerConfig: {},
  };
  },
- inject: [],
+ inject: [ConfigService],
  }),
  KafkaConsumerModule.forRootAsync({
- useFactory() {
+ useFactory: (configService: ConfigService) => {
  return {
  kafkaConfig: {
  clientId: 'kafka-client-id',
- brokers: ['localhost:9092'],
+ brokers: [configService.getOrThrow('KAFKA_BROKER')],
  logLevel: logLevel.NOTHING,
  },
  consumerConfig: { groupId: 'kafka-consumer-group' },
  };
  },
- inject: [],
+ inject: [ConfigService],
  }),
+ ConsumerModule,
  ],
  controllers: [],
  providers: [],

apps/service-auth/src/consumer/consumer.module.ts

@@ -0,0 +1,7 @@
+import { Module } from '@nestjs/common';
+import { ConsumerService } from './consumer.service';
+
+@Module({
+ providers: [ConsumerService],
+})
+export class ConsumerModule {}

apps/service-auth/src/consumer/consumer.service.ts

@@ -0,0 +1,28 @@
+import { KAFKA_CONSUMER } from '@app/kafka';
+import { Inject, Injectable } from '@nestjs/common';
+import { Consumer } from 'kafkajs';
+
+enum LogType {
+ CLAN_LEVEL_CHANGE = 'clan_level_change',
+ CLAN_WAR_LEAGUE_CHANGE = 'clan_war_league_change',
+ CAPITAL_LEAGUE_CHANGE = 'capital_league_change',
+ CLAN_MEMBER_CHANGE = 'clan_member_change',
+}
+
+@Injectable()
+export class ConsumerService {
+ constructor(@Inject(KAFKA_CONSUMER) private consumer: Consumer) {}
+
+ async onModuleInit() {
+ this.consumer.subscribe({ topics: Object.values(LogType), fromBeginning: true });
+
+ await this.consumer.run({
+ eachMessage: async ({ message, topic }) => {
+ console.log({
+ topic,
+ value: message.value?.toString(),
+ });
+ },
+ });
+ }
+}

apps/service-clans/src/service-clans.module.ts

@@ -1,12 +1,12 @@
 import { ClashClientModule } from '@app/clash-client';
+import { KafkaProducerModule } from '@app/kafka';
 import { MongoDbModule } from '@app/mongodb';
 import { RedisModule } from '@app/redis';
 import { Module } from '@nestjs/common';
-import { ConfigModule } from '@nestjs/config';
+import { ConfigModule, ConfigService } from '@nestjs/config';
+import { logLevel } from 'kafkajs';
 import { ServiceClansController } from './service-clans.controller';
 import { ClansService } from './service-clans.service';
-import { KafkaProducerModule } from '@app/kafka';
-import { logLevel } from 'kafkajs';
 
 @Module({
  imports: [
@@ -15,17 +15,17 @@ import { logLevel } from 'kafkajs';
  RedisModule,
  ClashClientModule,
  KafkaProducerModule.forRootAsync({
- useFactory() {
+ useFactory: (configService: ConfigService) => {
  return {
  kafkaConfig: {
  clientId: 'kafka-client-id',
- brokers: ['localhost:9092'],
+ brokers: [configService.getOrThrow('KAFKA_BROKER')],
  logLevel: logLevel.NOTHING,
  },
  producerConfig: {},
  };
  },
- inject: [],
+ inject: [ConfigService],
  }),
  ],
  controllers: [ServiceClansController],

libs/mongodb/src/mongodb.service.ts

@@ -22,6 +22,7 @@ export class MongoDbService {
  .aggregate<TrackedClanList>([
  {
  $match: {
+ active: true,
  paused: false,
  },
  },