Skip to content

Commit ced5ea4

Browse files
MarchosiaxMarchosiax
committed
Merge remote-tracking branch 'origin/dev' into dev
2 parents df435c0 + a530a63 commit ced5ea4

19 files changed

Lines changed: 233 additions & 63 deletions

File tree

‎accountant/accountant-core/src/main/kotlin/co/nilin/opex/accountant/core/model/KycLevel.kt‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
package co.nilin.opex.accountant.core.model
22

33
enum class KycLevel {
4-
Level1, Level2,
4+
LEVEL_1, LEVEL_2,LEVEL_3
55
}
66

‎auth-gateway/auth-gateway-app/src/main/kotlin/co/nilin/opex/auth/config/AppConfig.kt‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,12 @@
11
package co.nilin.opex.auth.config
22

3+
import co.nilin.opex.auth.kafka.KycLevelUpdatedKafkaListener
4+
import co.nilin.opex.auth.spi.KycLevelUpdatedEventListener
35
import jakarta.annotation.PostConstruct
46
import org.springframework.context.annotation.Configuration
57
import org.bouncycastle.util.io.pem.PemObject
68
import org.bouncycastle.util.io.pem.PemWriter
9+
import org.springframework.beans.factory.annotation.Autowired
710
import org.springframework.context.annotation.Bean
811
import java.io.File
912
import java.io.FileOutputStream
@@ -66,4 +69,13 @@ class AppConfig {
6669
PemWriter(stringWriter).use { it.writeObject(pemObject) }
6770
return stringWriter.toString()
6871
}
72+
73+
@Autowired
74+
fun configureEventListeners(
75+
kycLevelUpdatedKafkaListener: KycLevelUpdatedKafkaListener,
76+
kycLevelUpdatedEventListener: KycLevelUpdatedEventListener
77+
) {
78+
kycLevelUpdatedKafkaListener.addEventListener(kycLevelUpdatedEventListener)
79+
80+
}
6981
}

‎auth-gateway/auth-gateway-app/src/main/kotlin/co/nilin/opex/auth/config/KafkaListenerConfig.kt‎

Lines changed: 82 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,82 @@
1+
package co.nilin.opex.auth.config
2+
3+
4+
import co.nilin.opex.auth.data.KycLevelUpdatedEvent
5+
import co.nilin.opex.auth.kafka.KycLevelUpdatedKafkaListener
6+
import org.apache.kafka.clients.consumer.ConsumerConfig
7+
import org.apache.kafka.common.TopicPartition
8+
import org.apache.kafka.common.serialization.StringDeserializer
9+
import org.slf4j.LoggerFactory
10+
import org.springframework.beans.factory.annotation.Autowired
11+
import org.springframework.beans.factory.annotation.Qualifier
12+
import org.springframework.beans.factory.annotation.Value
13+
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean
14+
import org.springframework.context.annotation.Bean
15+
import org.springframework.context.annotation.Configuration
16+
import org.springframework.kafka.core.ConsumerFactory
17+
import org.springframework.kafka.core.DefaultKafkaConsumerFactory
18+
import org.springframework.kafka.core.DefaultKafkaProducerFactory
19+
import org.springframework.kafka.core.KafkaTemplate
20+
import org.springframework.kafka.listener.*
21+
import org.springframework.kafka.support.serializer.JsonDeserializer
22+
import org.springframework.util.backoff.FixedBackOff
23+
import java.util.regex.Pattern
24+
25+
@Configuration
26+
class KafkaListenerConfig {
27+
private val logger = LoggerFactory.getLogger(KafkaListenerConfig::class.java)
28+
29+
@Value("\${spring.kafka.bootstrap-servers}")
30+
private lateinit var bootstrapServers: String
31+
32+
@Value("\${spring.kafka.consumer.group-id}")
33+
private lateinit var groupId: String
34+
35+
@Bean("consumerConfigs")
36+
fun consumerConfigs(): Map<String, Any?> {
37+
38+
return mapOf(
39+
ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG to bootstrapServers,
40+
ConsumerConfig.GROUP_ID_CONFIG to groupId,
41+
ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG to StringDeserializer::class.java,
42+
ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG to JsonDeserializer::class.java,
43+
JsonDeserializer.TRUSTED_PACKAGES to "co.nilin.opex.*",
44+
JsonDeserializer.TYPE_MAPPINGS to "kyc_level_updated_event:co.nilin.opex.auth.data.KycLevelUpdatedEvent"
45+
)
46+
}
47+
48+
@Bean("kycLevelKafkaTemplate")
49+
fun kafkaTemplate(consumerConfigs: Map<String, Any?>): KafkaTemplate<String, KycLevelUpdatedEvent> {
50+
return KafkaTemplate(DefaultKafkaProducerFactory(consumerConfigs))
51+
}
52+
53+
@Bean
54+
fun kycConsumerFactory(@Qualifier("consumerConfigs") consumerConfigs: Map<String, Any?>): ConsumerFactory<String, KycLevelUpdatedEvent> {
55+
return DefaultKafkaConsumerFactory(consumerConfigs)
56+
}
57+
58+
@Autowired
59+
@Bean
60+
fun kycLevelUpdatedListenerContainer(
61+
listener: KycLevelUpdatedKafkaListener,
62+
consumerFactory: ConsumerFactory<String, KycLevelUpdatedEvent>,
63+
@Qualifier("kycLevelKafkaTemplate") template: KafkaTemplate<String, KycLevelUpdatedEvent>
64+
): ConcurrentMessageListenerContainer<String, KycLevelUpdatedEvent> {
65+
val containerProps = ContainerProperties(Pattern.compile("kyc_level_updated"))
66+
containerProps.messageListener = listener
67+
val container = ConcurrentMessageListenerContainer(consumerFactory, containerProps)
68+
container.setBeanName("KycLevelUpdatedKafkaListenerContainer")
69+
container.commonErrorHandler = createConsumerErrorHandler(template, "kyc_level_updated.DLT")
70+
container.start()
71+
return container
72+
}
73+
74+
private fun createConsumerErrorHandler(kafkaTemplate: KafkaTemplate<*, *>, dltTopic: String): CommonErrorHandler {
75+
val recoverer = DeadLetterPublishingRecoverer(kafkaTemplate) { cr, _ ->
76+
cr.headers().add("dlt-origin-module", "PROFILE".toByteArray())
77+
TopicPartition(dltTopic, cr.partition())
78+
}
79+
return DefaultErrorHandler(recoverer, FixedBackOff(5_000, 20))
80+
}
81+
82+
}

‎auth-gateway/auth-gateway-app/src/main/kotlin/co/nilin/opex/auth/data/KycLevelUpdatedEvent.kt‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,9 @@
1+
package co.nilin.opex.auth.data
2+
3+
import java.time.LocalDateTime
4+
5+
data class KycLevelUpdatedEvent(var userId: String, var kycLevel: KycLevel, var updateDate: LocalDateTime)
6+
7+
enum class KycLevel {
8+
LEVEL_1, LEVEL_2, LEVEL_3
9+
}

‎auth-gateway/auth-gateway-app/src/main/kotlin/co/nilin/opex/auth/data/UserRole.kt‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,7 @@
1+
package co.nilin.opex.auth.data
2+
3+
enum class UserRole(val keycloakName: String) {
4+
LEVEL_1("user-1"),
5+
LEVEL_2("user-2"),
6+
LEVEL_3("user-3"),
7+
}

‎auth-gateway/auth-gateway-app/src/main/kotlin/co/nilin/opex/auth/kafka/KycLevelUpdatedKafkaListener.kt‎

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,26 @@
1+
package co.nilin.opex.auth.kafka
2+
3+
4+
import co.nilin.opex.auth.data.KycLevelUpdatedEvent
5+
import co.nilin.opex.auth.spi.KycLevelUpdatedEventListener
6+
import org.apache.kafka.clients.consumer.ConsumerRecord
7+
import org.slf4j.LoggerFactory
8+
import org.springframework.kafka.listener.MessageListener
9+
import org.springframework.stereotype.Component
10+
11+
@Component
12+
class KycLevelUpdatedKafkaListener : MessageListener<String, KycLevelUpdatedEvent> {
13+
val eventListeners = arrayListOf<KycLevelUpdatedEventListener>()
14+
private val logger = LoggerFactory.getLogger(KycLevelUpdatedKafkaListener::class.java)
15+
override fun onMessage(data: ConsumerRecord<String, KycLevelUpdatedEvent>) {
16+
17+
eventListeners.forEach { tl ->
18+
logger.info("incoming new event " + tl.id())
19+
tl.onEvent(data.value(), data.partition(), data.offset(), data.timestamp(), tl.id())
20+
}
21+
}
22+
23+
fun addEventListener(tl: KycLevelUpdatedEventListener) {
24+
eventListeners.add(tl)
25+
}
26+
}

‎auth-gateway/auth-gateway-app/src/main/kotlin/co/nilin/opex/auth/kafka/KycLevelUpdatedListener.kt‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,32 @@
1+
package co.nilin.opex.auth.kafka
2+
3+
import co.nilin.opex.auth.data.KycLevelUpdatedEvent
4+
import co.nilin.opex.auth.data.UserRole
5+
import co.nilin.opex.auth.proxy.KeycloakProxy
6+
import co.nilin.opex.auth.spi.KycLevelUpdatedEventListener
7+
import kotlinx.coroutines.CoroutineScope
8+
import kotlinx.coroutines.Dispatchers
9+
import kotlinx.coroutines.launch
10+
import org.slf4j.LoggerFactory
11+
import org.springframework.stereotype.Component
12+
13+
@Component
14+
class KycLevelUpdatedListener(private val keycloakProxy: KeycloakProxy) : KycLevelUpdatedEventListener {
15+
16+
private val logger = LoggerFactory.getLogger(KycLevelUpdatedListener::class.java)
17+
val scope = CoroutineScope(Dispatchers.IO)
18+
override fun id(): String {
19+
return "KycLevelUpdatedListener"
20+
}
21+
22+
override fun onEvent(
23+
event: KycLevelUpdatedEvent, partition: Int, offset: Long, timestamp: Long, eventId: String
24+
) {
25+
logger.info("==========================================================================")
26+
logger.info("Incoming UserLevelUpdated event: $event")
27+
logger.info("==========================================================================")
28+
scope.launch {
29+
keycloakProxy.assignRole(event.userId, UserRole.valueOf(event.kycLevel.name))
30+
}
31+
}
32+
}

‎auth-gateway/auth-gateway-app/src/main/kotlin/co/nilin/opex/auth/proxy/KeycloakProxy.kt‎

Lines changed: 28 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@ package co.nilin.opex.auth.proxy
22

33
import co.nilin.opex.auth.config.KeycloakConfig
44
import co.nilin.opex.auth.data.ActiveSession
5+
import co.nilin.opex.auth.data.UserRole
56
import co.nilin.opex.auth.model.*
67
import co.nilin.opex.auth.utils.generateRandomID
78
import co.nilin.opex.common.OpexError
@@ -211,10 +212,10 @@ class KeycloakProxy(
211212
.awaitSingle()
212213
}
213214

214-
suspend fun assignDefaultRoles(user: KeycloakUser) {
215-
val role = opexRealm.roles().get("user-1").toRepresentation()
216-
val u = opexRealm.users().get(user.id)
217-
u.roles().realmLevel().add(mutableListOf(role))
215+
suspend fun assignRole(userId: String, role: UserRole) {
216+
val roleRepresentation = opexRealm.roles().get(role.keycloakName).toRepresentation()
217+
val userResource = opexRealm.users().get(userId)
218+
userResource.roles().realmLevel().add(listOf(roleRepresentation))
218219
}
219220

220221
suspend fun createExternalIdpUser(email: String, username: Username, password: String): String {
@@ -363,53 +364,47 @@ class KeycloakProxy(
363364
}
364365

365366
suspend fun updateUserMobile(userId: String, newMobile: String) {
366-
val url = "${keycloakConfig.url}/admin/realms/${keycloakConfig.realm}/users/$userId"
367-
val patch = mapOf(
368-
"attributes" to mapOf("mobile" to newMobile)
369-
)
370-
keycloakClient.put()
371-
.uri(url)
372-
.contentType(MediaType.APPLICATION_JSON)
373-
.withAdminToken()
374-
.bodyValue(patch)
375-
.retrieve()
376-
.toBodilessEntity()
377-
.awaitSingleOrNull()
367+
updateUserFields(userId, mapOf("attributes" to mapOf("mobile" to newMobile)))
378368
}
379369

380370
suspend fun updateUserEmail(userId: String, newEmail: String) {
381-
val url = "${keycloakConfig.url}/admin/realms/${keycloakConfig.realm}/users/$userId"
382-
val patch = mapOf(
383-
"email" to newEmail,
384-
"emailVerified" to true
385-
)
386-
keycloakClient.put()
387-
.uri(url)
388-
.contentType(MediaType.APPLICATION_JSON)
389-
.withAdminToken()
390-
.bodyValue(patch)
391-
.retrieve()
392-
.toBodilessEntity()
393-
.awaitSingleOrNull()
371+
updateUserFields(userId, mapOf("email" to newEmail, "emailVerified" to true))
394372
}
395373

396374
suspend fun updateUserName(userId: String, firstName: String?, lastName: String?) {
375+
val updates = mutableMapOf<String, Any>()
376+
firstName?.let { updates["firstName"] = it }
377+
lastName?.let { updates["lastName"] = it }
378+
updateUserFields(userId, updates)
379+
380+
}
381+
382+
private suspend fun updateUserFields(userId: String, updates: Map<String, Any>) {
397383
val url = "${keycloakConfig.url}/admin/realms/${keycloakConfig.realm}/users/$userId"
398-
val patch = mutableMapOf<String, Any>()
399384

400-
firstName?.let { patch["firstName"] = it }
401-
lastName?.let { patch["lastName"] = it }
385+
val existingUser = keycloakClient.get()
386+
.uri(url)
387+
.withAdminToken()
388+
.retrieve()
389+
.bodyToMono<Map<String, Any>>()
390+
.awaitSingle()
391+
.toMutableMap()
392+
393+
updates.forEach { (key, value) ->
394+
existingUser[key] = value
395+
}
402396

403397
keycloakClient.put()
404398
.uri(url)
405399
.contentType(MediaType.APPLICATION_JSON)
406400
.withAdminToken()
407-
.bodyValue(patch)
401+
.bodyValue(existingUser)
408402
.retrieve()
409403
.toBodilessEntity()
410404
.awaitSingleOrNull()
411405
}
412406

407+
413408
private suspend fun generateRandomInternalID(): String {
414409
var internalId: String;
415410
var attempts = 0

‎auth-gateway/auth-gateway-app/src/main/kotlin/co/nilin/opex/auth/service/UserService.kt‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ package co.nilin.opex.auth.service
33
import co.nilin.opex.auth.data.ActionType
44
import co.nilin.opex.auth.data.ActiveSession
55
import co.nilin.opex.auth.data.UserCreatedEvent
6+
import co.nilin.opex.auth.data.UserRole
67
import co.nilin.opex.auth.kafka.AuthEventProducer
78
import co.nilin.opex.auth.model.*
89
import co.nilin.opex.auth.proxy.GoogleProxy
@@ -80,7 +81,7 @@ class UserService(
8081
throw OpexError.BadRequest.exception()
8182

8283
keycloakProxy.confirmCreateUser(user, request.password)
83-
keycloakProxy.assignDefaultRoles(user)
84+
keycloakProxy.assignRole(user.id, UserRole.LEVEL_1)
8485

8586
// Send event to let other services know a user just registered
8687
val event = UserCreatedEvent(user.id, user.username, user.email, user.mobile, user.firstName, user.lastName)
@@ -214,5 +215,5 @@ class UserService(
214215
return TokenData(false, "", OTPAction.REGISTER)
215216
}
216217
}
217-
218+
218219
}

‎auth-gateway/auth-gateway-app/src/main/kotlin/co/nilin/opex/auth/spi/KycLevelUpdatedEventListener.kt‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,10 @@
1+
package co.nilin.opex.auth.spi
2+
3+
import co.nilin.opex.auth.data.KycLevelUpdatedEvent
4+
5+
6+
interface KycLevelUpdatedEventListener {
7+
fun id(): String
8+
fun onEvent(event: KycLevelUpdatedEvent, partition: Int, offset: Long, timestamp: Long, eventId: String)
9+
10+
}

0 commit comments

Comments
 (0)