feat: send telegram messages (#43)

* feat: send telegram messages

task: #12

* fix code format

* add test cases and correct service

* remove default chat_id

* Minor refact in Telegram integration

* Update docs for Telegram integration

Co-authored-by: akobor <[email protected]>

README.md

@@ -24,6 +24,7 @@ Kuvasz (pronounce as [ˈkuvɒs]) is an ancient hungarian breed of livestock & gu
 - Uptime & latency monitoring with a configurable interval
 - Email notifications through SMTP
 - Slack notifications through webhoooks
+- Telegram notifications through the Bot API
 - Configurable data retention period
 
 ### Under development 🚧

examples/docker-compose/docker-compose.yml

@@ -28,3 +28,6 @@ services:
       ENABLE_SLACK_EVENT_HANDLER: 'true'
       SLACK_WEBHOOK_URL: 'https://your.slack-webhook.url'
       DATA_RETENTION_DAYS: 30
+      ENABLE_TELEGRAM_EVENT_HANDLER: 'true'
+      TELEGRAM_API_TOKEN: '1232312321321:GJKGHjhklfdhsklHKLFH'
+      TELEGRAM_CHAT_ID: '1234567890'

examples/k8s/kuvasz.configmap.yml

@@ -18,3 +18,4 @@ data:
   slack_event_handler_enabled: "true"
   slack_webhook_url: "https://your.slack-webhook.url"
   data_retention_days: "30"
+  telegram_event_handler_enabled: "true"

examples/k8s/kuvasz.deployment.yaml

@@ -132,3 +132,18 @@ spec:
                 configMapKeyRef:
                   name: kuvasz-config
                   key: data_retention_days
+            - name: ENABLE_TELEGRAM_EVENT_HANDLER
+              valueFrom:
+                configMapKeyRef:
+                  name: kuvasz-config
+                  key: telegram_event_handler_enabled
+            - name: TELEGRAM_API_TOKEN
+              valueFrom:
+                secretKeyRef:
+                  name: telegram-credentials
+                  key: api-token
+            - name: TELEGRAM_CHAT_ID
+              valueFrom:
+                secretKeyRef:
+                  name: telegram-credentials
+                  key: chat-id

src/main/kotlin/com/kuvaszuptime/kuvasz/config/handlers/TelegramEventHandlerConfig.kt

@@ -0,0 +1,18 @@
+package com.kuvaszuptime.kuvasz.config.handlers
+
+import io.micronaut.context.annotation.ConfigurationProperties
+import io.micronaut.core.annotation.Introspected
+import javax.inject.Singleton
+import javax.validation.constraints.NotBlank
+
+@ConfigurationProperties("handler-config.telegram-event-handler")
+@Singleton
+@Introspected
+class TelegramEventHandlerConfig {
+
+    @NotBlank
+    var token: String = ""
+
+    @NotBlank
+    var chatId: String = ""
+}

src/main/kotlin/com/kuvaszuptime/kuvasz/handlers/TelegramEventHandler.kt

@@ -0,0 +1,84 @@
+package com.kuvaszuptime.kuvasz.handlers
+
+import com.kuvaszuptime.kuvasz.config.handlers.TelegramEventHandlerConfig
+import com.kuvaszuptime.kuvasz.models.MonitorDownEvent
+import com.kuvaszuptime.kuvasz.models.MonitorUpEvent
+import com.kuvaszuptime.kuvasz.models.TelegramAPIMessage
+import com.kuvaszuptime.kuvasz.models.UptimeMonitorEvent
+import com.kuvaszuptime.kuvasz.models.runWhenStateChanges
+import com.kuvaszuptime.kuvasz.models.toEmoji
+import com.kuvaszuptime.kuvasz.models.toStructuredMessage
+import com.kuvaszuptime.kuvasz.services.EventDispatcher
+import com.kuvaszuptime.kuvasz.services.TelegramAPIService
+import io.micronaut.context.annotation.Context
+import io.micronaut.context.annotation.Requires
+import io.micronaut.http.HttpResponse
+import io.micronaut.http.client.exceptions.HttpClientResponseException
+import io.micronaut.scheduling.TaskExecutors
+import io.micronaut.scheduling.annotation.ExecuteOn
+import io.reactivex.Flowable
+import org.slf4j.LoggerFactory
+
+@Context
+@Requires(property = "handler-config.telegram-event-handler.enabled", value = "true")
+class TelegramEventHandler(
+    private val telegramAPIService: TelegramAPIService,
+    private val telegramEventHandlerConfig: TelegramEventHandlerConfig,
+    private val eventDispatcher: EventDispatcher
+) {
+    companion object {
+        private val logger = LoggerFactory.getLogger(TelegramEventHandler::class.java)
+    }
+
+    init {
+        subscribeToEvents()
+    }
+
+    @ExecuteOn(TaskExecutors.IO)
+    private fun subscribeToEvents() {
+        eventDispatcher.subscribeToMonitorUpEvents { event ->
+            logger.debug("A MonitorUpEvent has been received for monitor with ID: ${event.monitor.id}")
+            event.runWhenStateChanges { telegramAPIService.sendMessage(it.toTelegramMessage()).handleResponse() }
+        }
+        eventDispatcher.subscribeToMonitorDownEvents { event ->
+            logger.debug("A MonitorDownEvent has been received for monitor with ID: ${event.monitor.id}")
+            event.runWhenStateChanges { telegramAPIService.sendMessage(it.toTelegramMessage()).handleResponse() }
+        }
+    }
+
+    private fun UptimeMonitorEvent.toTelegramMessage(): TelegramAPIMessage =
+        TelegramAPIMessage(
+            text = "${toEmoji()} ${toHTMLMessage()}",
+            chat_id = telegramEventHandlerConfig.chatId
+        )
+
+    private fun Flowable<HttpResponse<String>>.handleResponse() =
+        subscribe(
+            {
+                logger.debug("A Telegram message to your configured webhook has been successfully sent")
+            },
+            { ex ->
+                if (ex is HttpClientResponseException) {
+                    val responseBody = ex.response.getBody(String::class.java)
+                    logger.error("Telegram message cannot be delivered due to an error: $responseBody")
+                }
+            }
+        )
+
+    private fun UptimeMonitorEvent.toHTMLMessage() =
+        when (this) {
+            is MonitorUpEvent -> toStructuredMessage().let { details ->
+                listOfNotNull(
+                    "<b>${details.summary}</b>",
+                    "<i>${details.latency}</i>",
+                    details.previousDownTime.orNull()
+                )
+            }
+            is MonitorDownEvent -> toStructuredMessage().let { details ->
+                listOfNotNull(
+                    "<b>${details.summary}</b>",
+                    details.previousUpTime.orNull()
+                )
+            }
+        }.joinToString("\n")
+}

src/main/kotlin/com/kuvaszuptime/kuvasz/models/TelegramAPIMessage.kt

@@ -0,0 +1,12 @@
+package com.kuvaszuptime.kuvasz.models
+
+import io.micronaut.core.annotation.Introspected
+
+@Suppress("ConstructorParameterNaming")
+@Introspected
+data class TelegramAPIMessage(
+    val chat_id: String,
+    val text: String,
+    val disable_web_page_preview: Boolean = true,
+    val parse_mode: String = "HTML"
+)

src/main/kotlin/com/kuvaszuptime/kuvasz/services/TelegramAPIService.kt

@@ -0,0 +1,41 @@
+package com.kuvaszuptime.kuvasz.services
+
+import com.kuvaszuptime.kuvasz.config.handlers.TelegramEventHandlerConfig
+import com.kuvaszuptime.kuvasz.models.TelegramAPIMessage
+import io.micronaut.context.annotation.Requires
+import io.micronaut.context.event.ShutdownEvent
+import io.micronaut.core.type.Argument
+import io.micronaut.http.HttpRequest
+import io.micronaut.http.HttpResponse
+import io.micronaut.http.client.RxHttpClient
+import io.micronaut.runtime.event.annotation.EventListener
+import io.reactivex.Flowable
+import javax.inject.Inject
+import javax.inject.Singleton
+
+@Singleton
+@Requires(property = "handler-config.telegram-event-handler.enabled", value = "true")
+class TelegramAPIService @Inject constructor(
+    telegramEventHandlerConfig: TelegramEventHandlerConfig,
+    private val httpClient: RxHttpClient
+) {
+    private val url = "https://api.telegram.org/bot" + telegramEventHandlerConfig.token + "/sendMessage"
+
+    companion object {
+        private const val RETRY_COUNT = 3L
+    }
+
+    fun sendMessage(message: TelegramAPIMessage): Flowable<HttpResponse<String>> {
+        val request: HttpRequest<TelegramAPIMessage> = HttpRequest.POST(url, message)
+
+        return httpClient
+            .exchange(request, Argument.STRING, Argument.STRING)
+            .retry(RETRY_COUNT)
+    }
+
+    @EventListener
+    @Suppress("UNUSED_PARAMETER")
+    internal fun onShutdownEvent(event: ShutdownEvent) {
+        httpClient.close()
+    }
+}

src/main/resources/application.yml

@@ -89,6 +89,10 @@ handler-config:
   slack-event-handler:
     enabled: ${ENABLE_SLACK_EVENT_HANDLER:`false`}
     webhook-url: ${SLACK_WEBHOOK_URL}
+  telegram-event-handler:
+    enabled: ${ENABLE_TELEGRAM_EVENT_HANDLER:`false`}
+    token: ${TELEGRAM_API_TOKEN}
+    chat-id: ${TELEGRAM_CHAT_ID}
 ---
 admin-auth:
   username: ${ADMIN_USER}

src/test/kotlin/com/kuvaszuptime/kuvasz/config/TelegramEventHandlerConfigTest.kt

@@ -0,0 +1,65 @@
+package com.kuvaszuptime.kuvasz.config
+
+import io.kotest.assertions.exceptionToMessage
+import io.kotest.assertions.throwables.shouldThrow
+import io.kotest.core.spec.style.BehaviorSpec
+import io.kotest.matchers.string.shouldContain
+import io.micronaut.context.ApplicationContext
+import io.micronaut.context.env.PropertySource
+import io.micronaut.context.exceptions.BeanInstantiationException
+
+class TelegramEventHandlerConfigTest : BehaviorSpec({
+    given("a TelegramEventHandlerConfig bean") {
+        `when`("there is no API token in the configuration") {
+            val properties = PropertySource.of(
+                "test",
+                mapOf(
+                    "handler-config.telegram-event-handler.enabled" to "true",
+                    "handler-config.telegram-event-handler.chat-id" to "chat-id"
+                )
+            )
+            then("ApplicationContext should throw a BeanInstantiationException") {
+                val exception = shouldThrow<BeanInstantiationException> {
+                    ApplicationContext.run(properties)
+                }
+                exceptionToMessage(exception) shouldContain
+                        "Bean definition [com.kuvaszuptime.kuvasz.handlers.TelegramEventHandler] could not be loaded"
+            }
+        }
+
+        `when`("there is no chat ID in the configuration") {
+            val properties = PropertySource.of(
+                "test",
+                mapOf(
+                    "handler-config.telegram-event-handler.enabled" to "true",
+                    "handler-config.telegram-event-handler.token" to "your-token"
+                )
+            )
+            then("ApplicationContext should throw a BeanInstantiationException") {
+                val exception = shouldThrow<BeanInstantiationException> {
+                    ApplicationContext.run(properties)
+                }
+                exceptionToMessage(exception) shouldContain
+                        "Bean definition [com.kuvaszuptime.kuvasz.handlers.TelegramEventHandler] could not be loaded"
+            }
+        }
+
+        `when`("chat ID and API token are empty strings") {
+            val properties = PropertySource.of(
+                "test",
+                mapOf(
+                    "handler-config.telegram-event-handler.enabled" to "true",
+                    "handler-config.telegram-event-handler.token" to "",
+                    "handler-config.telegram-event-handler.chat-id" to ""
+                )
+            )
+            then("ApplicationContext should throw a BeanInstantiationException") {
+                val exception = shouldThrow<BeanInstantiationException> {
+                    ApplicationContext.run(properties)
+                }
+                exceptionToMessage(exception) shouldContain
+                        "Bean definition [com.kuvaszuptime.kuvasz.handlers.TelegramEventHandler] could not be loaded"
+            }
+        }
+    }
+})