#139 Send publications to admin via telegram (#145)

botalka/build.gradle.kts

@@ -23,6 +23,8 @@ dependencies {
 
     implementation("io.projectreactor.kafka:reactor-kafka")
     testImplementation("org.testcontainers:kafka")
+
+    implementation("io.github.kotlin-telegram-bot.kotlin-telegram-bot:telegram:6.1.0")
 }
 
 val jooqPackageName = "$basePackage.storage.jooq"

botalka/src/main/kotlin/ru/vityaman/lms/botalka/app/spring/api/http/endpoint/RatingHttpApi.kt

@@ -19,7 +19,6 @@ class RatingHttpApi(
         ResponseEntity.ok(
             RatingGrades(
                 grades = service.grades().map { (student, grades) ->
-                    println("$student $grades")
                     StudentGradesMessage(
                         studentId = student.id.number,
                         grades = grades.entries

botalka/src/main/kotlin/ru/vityaman/lms/botalka/app/spring/client/telegram/SpringTelegramBot.kt

@@ -0,0 +1,39 @@
+package ru.vityaman.lms.botalka.app.spring.client.telegram
+
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.runBlocking
+import org.springframework.beans.factory.annotation.Value
+import org.springframework.boot.context.event.ApplicationReadyEvent
+import org.springframework.context.annotation.Profile
+import org.springframework.context.event.EventListener
+import org.springframework.stereotype.Service
+import ru.vityaman.lms.botalka.core.external.telegram.BasicTelegramBot
+import ru.vityaman.lms.botalka.core.external.telegram.TelegramBot
+import ru.vityaman.lms.botalka.core.external.telegram.TelegramChat
+import ru.vityaman.lms.botalka.core.external.telegram.TelegramMessage
+
+@Service
+@Profile("production")
+class SpringTelegramBot(
+    @Value("\${external.service.telegram.token}")
+    token: String,
+
+    @Value("\${external.service.telegram.admin-chat-id}")
+    adminChatId: Long,
+) : TelegramBot by BasicTelegramBot(token) {
+    private val scope = CoroutineScope(Dispatchers.Default)
+    private val adminChat = TelegramChat(adminChatId)
+
+    init {
+        scope.launch {
+            start()
+        }
+    }
+
+    @EventListener(ApplicationReadyEvent::class)
+    fun notifyAdmin() = runBlocking {
+        send(adminChat, TelegramMessage("Restarted instance"))
+    }
+}

botalka/src/main/kotlin/ru/vityaman/lms/botalka/app/spring/task/SpringPublicationConsumer.kt → botalka/src/main/kotlin/ru/vityaman/lms/botalka/app/spring/task/SpringKafkaPublicationConsumer.kt

@@ -10,7 +10,8 @@ import ru.vityaman.lms.botalka.storage.kafka.KafkaProducer
 import ru.vityaman.lms.botalka.storage.kafka.KafkaTopic
 
 @Component
-class SpringPublicationConsumer(
+@Qualifier(SpringPublicationConfig.BeanName.KAFKA_CONSUMER)
+class SpringKafkaPublicationConsumer(
     @Value("\${broker.bootstrap-servers}")
     bootstrapServers: String,
 

botalka/src/main/kotlin/ru/vityaman/lms/botalka/app/spring/task/SpringNotificationTask.kt

@@ -1,12 +1,12 @@
 package ru.vityaman.lms.botalka.app.spring.task
 
 import kotlinx.coroutines.runBlocking
+import org.springframework.beans.factory.annotation.Qualifier
 import org.springframework.beans.factory.annotation.Value
 import org.springframework.scheduling.annotation.Scheduled
 import org.springframework.stereotype.Component
-import ru.vityaman.lms.botalka.commons.Consumer
 import ru.vityaman.lms.botalka.core.logging.Slf4jLog
-import ru.vityaman.lms.botalka.core.model.Homework
+import ru.vityaman.lms.botalka.core.publication.PublicationConsumer
 import ru.vityaman.lms.botalka.core.publication.PublicationSupplier
 import ru.vityaman.lms.botalka.core.publication.logging.loggingNotificationCallbacks
 import ru.vityaman.lms.botalka.core.publication.task.NotificationTask
@@ -19,20 +19,17 @@ class SpringNotificationTask(
     batchDurationSeconds: Int,
 
     supplier: PublicationSupplier,
-) {
-    private val log = Slf4jLog("NotificationTask")
 
+    @Qualifier(SpringPublicationConfig.BeanName.TELEGRAM_CONSUMER)
+    consumer: PublicationConsumer,
+) {
     private val logic = NotificationTask(
         supplier = supplier,
-        consumer = object : Consumer<Homework> {
-            override suspend fun accept(value: Homework) {
-                log.info("Consumed $value")
-            }
-        },
+        consumer = consumer,
         config = NotificationTask.Config(
             batchDuration = batchDurationSeconds.seconds,
         ),
-        callbacks = loggingNotificationCallbacks(log),
+        callbacks = loggingNotificationCallbacks(Slf4jLog("NotificationTask")),
     )
 
     @Scheduled(

botalka/src/main/kotlin/ru/vityaman/lms/botalka/app/spring/task/SpringPublicationConfig.kt

@@ -0,0 +1,8 @@
+package ru.vityaman.lms.botalka.app.spring.task
+
+class SpringPublicationConfig {
+    object BeanName {
+        const val KAFKA_CONSUMER = "kafkaPublicationConsumer"
+        const val TELEGRAM_CONSUMER = "telegramPublicationConsumer"
+    }
+}

botalka/src/main/kotlin/ru/vityaman/lms/botalka/app/spring/task/SpringPublicationTask.kt

@@ -18,6 +18,7 @@ import java.util.concurrent.TimeUnit
 class SpringPublicationTask(
     homeworks: HomeworkStorage,
 
+    @Qualifier(SpringPublicationConfig.BeanName.KAFKA_CONSUMER)
     consumer: PublicationConsumer,
 
     @Qualifier(MainR2dbcConfig.BeanName.TX_ENV)

botalka/src/main/kotlin/ru/vityaman/lms/botalka/app/spring/task/SpringTelegramPublicationConsumer.kt

@@ -0,0 +1,19 @@
+package ru.vityaman.lms.botalka.app.spring.task
+
+import org.springframework.beans.factory.annotation.Qualifier
+import org.springframework.beans.factory.annotation.Value
+import org.springframework.stereotype.Component
+import ru.vityaman.lms.botalka.core.external.telegram.TelegramBot
+import ru.vityaman.lms.botalka.core.external.telegram.TelegramChat
+import ru.vityaman.lms.botalka.core.publication.PublicationConsumer
+import ru.vityaman.lms.botalka.core.publication.telegram.TelegramPublicationConsumer
+
+@Component
+@Qualifier(SpringPublicationConfig.BeanName.TELEGRAM_CONSUMER)
+class SpringTelegramPublicationConsumer(
+    telegram: TelegramBot,
+
+    @Value("\${external.service.telegram.admin-chat-id}")
+    adminChatId: Long,
+) : PublicationConsumer by
+TelegramPublicationConsumer(telegram, TelegramChat(adminChatId))

botalka/src/main/kotlin/ru/vityaman/lms/botalka/core/external/telegram/BasicTelegramBot.kt

@@ -0,0 +1,42 @@
+package ru.vityaman.lms.botalka.core.external.telegram
+
+import com.github.kotlintelegrambot.bot
+import com.github.kotlintelegrambot.dispatch
+import com.github.kotlintelegrambot.dispatcher.text
+import com.github.kotlintelegrambot.entities.ChatId
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.withContext
+
+class BasicTelegramBot(accessToken: String) : TelegramBot {
+    private val bot = bot {
+        token = accessToken
+        dispatch {
+            text {
+                send(
+                    TelegramChat(message.chat.id),
+                    TelegramMessage(
+                        text = mapOf(
+                            "Chat" to message.chat.id,
+                            "From" to message.from?.id,
+                            "Text" to message.text,
+                        ).entries.joinToString(", ") {
+                            "${it.key}: ${it.value}"
+                        },
+                    ),
+                )
+            }
+        }
+    }
+
+    override suspend fun send(chat: TelegramChat, message: TelegramMessage) {
+        withContext(Dispatchers.IO) {
+            bot.sendMessage(ChatId.fromId(chat.id), text = message.text).get()
+        }
+    }
+
+    override suspend fun start() {
+        withContext(Dispatchers.IO) {
+            bot.startPolling()
+        }
+    }
+}

botalka/src/main/kotlin/ru/vityaman/lms/botalka/core/external/telegram/TelegramBot.kt

@@ -0,0 +1,6 @@
+package ru.vityaman.lms.botalka.core.external.telegram
+
+interface TelegramBot {
+    suspend fun send(chat: TelegramChat, message: TelegramMessage)
+    suspend fun start()
+}

botalka/src/main/kotlin/ru/vityaman/lms/botalka/core/external/telegram/TelegramChat.kt

@@ -0,0 +1,3 @@
+package ru.vityaman.lms.botalka.core.external.telegram
+
+data class TelegramChat(val id: Long)

botalka/src/main/kotlin/ru/vityaman/lms/botalka/core/external/telegram/TelegramMessage.kt

@@ -0,0 +1,3 @@
+package ru.vityaman.lms.botalka.core.external.telegram
+
+data class TelegramMessage(val text: String)

botalka/src/main/kotlin/ru/vityaman/lms/botalka/core/publication/telegram/TelegramPublicationConsumer.kt

@@ -0,0 +1,28 @@
+package ru.vityaman.lms.botalka.core.publication.telegram
+
+import kotlinx.coroutines.delay
+import ru.vityaman.lms.botalka.core.external.telegram.TelegramBot
+import ru.vityaman.lms.botalka.core.external.telegram.TelegramChat
+import ru.vityaman.lms.botalka.core.external.telegram.TelegramMessage
+import ru.vityaman.lms.botalka.core.model.Homework
+import ru.vityaman.lms.botalka.core.publication.PublicationConsumer
+import kotlin.time.Duration.Companion.seconds
+
+class TelegramPublicationConsumer(
+    private val telegram: TelegramBot,
+    private val adminChat: TelegramChat,
+) : PublicationConsumer {
+    override suspend fun accept(value: Homework) {
+        val text = buildString {
+            append("Published homework '${value.title.text}'!\n")
+            append("\n")
+            append("${value.description}\n")
+            append("\n")
+            append("MaxScore: ${value.maxScore.value}\n")
+            append("Deadline: ${value.deadlineMoment}\n")
+            append("Id: ${value.id.number}\n")
+        }
+        telegram.send(adminChat, TelegramMessage(text))
+        delay(1.seconds)
+    }
+}

botalka/src/main/kotlin/ru/vityaman/lms/botalka/storage/kafka/KafkaProducer.kt

@@ -34,7 +34,6 @@ class KafkaProducer<K, V>(
             .let { SenderRecord.create(it, null) }
             .let { mono { it } }
         kafka.send(record).asFlow().collect {}
-        println("Sent $value")
     }
 
     data class Config(

botalka/src/main/resources/application-test.yml

@@ -11,6 +11,9 @@ external:
         clientSecret: fake-yandex-client-secret
       login:
         url: http://localhost:8080/fake/external/service/yandex/login
+    telegram:
+      token: fake-telegram-bot-api-token
+      admin-chat-id: 0
 task:
   scheduled:
     publication:

botalka/src/main/resources/application.yml

@@ -1,4 +1,6 @@
 spring:
+  profiles:
+    active: production
   application:
     name: botalka
   datasource:
@@ -66,6 +68,9 @@ external:
         clientSecret: ${LMS_YANDEX_CLIENT_SECRET}
       login:
         url: https://login.yandex.ru
+    telegram:
+      token: ${LMS_TEST_TELEGRAM_BOT_API_TOKEN}
+      admin-chat-id: ${LMS_TEST_TELEGRAM_ADMIN_CHAT_ID}
 task:
   scheduled:
     publication:

botalka/src/test/kotlin/ru/vityaman/lms/botalka/app/spring/client/telegram/FakeTelegramBot.kt

@@ -0,0 +1,22 @@
+package ru.vityaman.lms.botalka.app.spring.client.telegram
+
+import org.slf4j.LoggerFactory
+import org.springframework.context.annotation.Primary
+import org.springframework.stereotype.Service
+import ru.vityaman.lms.botalka.core.external.telegram.TelegramBot
+import ru.vityaman.lms.botalka.core.external.telegram.TelegramChat
+import ru.vityaman.lms.botalka.core.external.telegram.TelegramMessage
+
+@Primary
+@Service
+class FakeTelegramBot : TelegramBot {
+    private val log = LoggerFactory.getLogger("FakeTelegramBot")
+
+    override suspend fun send(chat: TelegramChat, message: TelegramMessage) {
+        log.debug("Received a {} from {}", message, chat)
+    }
+
+    override suspend fun start() {
+        log.debug("Started")
+    }
+}

buildSrc/src/main/kotlin/lms.conventions.common.gradle.kts

@@ -9,4 +9,5 @@ version = "0.0.1"
 
 repositories {
     mavenCentral()
+    maven(url = "https://jitpack.io")
 }

compose.yml

@@ -25,6 +25,8 @@ services:
       LMS_YANDEX_CLIENT_ID: ${LMS_YANDEX_CLIENT_ID?:err}
       LMS_YANDEX_CLIENT_SECRET: ${LMS_YANDEX_CLIENT_SECRET?:err}
       LMS_SECURITY_TOKEN_SIGNING_SECRET: ${LMS_SECURITY_TOKEN_SIGNING_SECRET?:err}
+      LMS_TEST_TELEGRAM_BOT_API_TOKEN: ${LMS_TEST_TELEGRAM_BOT_API_TOKEN?:err}
+      LMS_TEST_TELEGRAM_ADMIN_CHAT_ID: ${LMS_TEST_TELEGRAM_ADMIN_CHAT_ID?:err}
     networks:
       - lms-network
     depends_on:

infra/down.sh

@@ -0,0 +1,13 @@
+#!/usr/bin/sh
+
+set -e
+
+cd "$(dirname "$0")"/..  || exit
+
+echo "[up] Setting up local variables..."
+. ./infra/env/local.sh
+
+echo "[up] Setting up secrets..."
+. ./infra/env/secret.sh
+
+docker compose down

infra/fuzzing/custom/homeworks.sh

@@ -20,5 +20,5 @@ for _ in $(seq 1 128); do
       \"max_score\": 500,
       \"publication_moment\": \"$MOMENT\",
       \"deadline_moment\": \"2026-07-11T12:00:00Z\"
-    }"
+    }" &
 done

infra/kafka/setup.sh

@@ -21,7 +21,7 @@ topic() {
   $KAFKA_TOPICS \
     --bootstrap-server "$HOST:$PORT" \
     --create --if-not-exists --topic "$NAME" \
-    --replication-factor 1 --partitions 1
+    --replication-factor 1 --partitions 2
 }
 
 topic "publication"

infra/env/up.sh → infra/up.sh

@@ -1,8 +1,8 @@
-#!/usr/bin/env sh
+#!/usr/bin/sh
 
 set -e
 
-cd "$(dirname "$0")"/../..  || exit
+cd "$(dirname "$0")"/..  || exit
 
 echo "[up] Setting up local variables..."
 . ./infra/env/local.sh