Implemented telegram notification

.gitignore

@@ -44,3 +44,5 @@ out/
 *.lock.db
 
 src/main/resources/banyuwangi-dashboard-firebase-adminsdk.json
+.env
+.env.*

build.gradle.kts

@@ -13,6 +13,7 @@ plugins {
 
     id("org.springdoc.openapi-gradle-plugin") version "1.8.0"
     id("org.gradle.test-retry") version "1.5.9"
+    id("co.uzzu.dotenv.gradle") version "4.0.0"
 }
 
 val buildId = System.getenv("GITHUB_RUN_NUMBER") ?: System.getenv("BUILD_ID") ?: "1-SNAPSHOT"
@@ -122,6 +123,8 @@ dependencies {
 
     // Firebase cloud messaging
     implementation("com.google.firebase:firebase-admin:9.1.1")
+
+    implementation("io.github.kotlin-telegram-bot.kotlin-telegram-bot:telegram:6.3.0")
 }
 
 tasks.withType<KotlinCompile> {
@@ -135,6 +138,7 @@ tasks.withType<Test> {
     useJUnitPlatform()
     enableAssertions = true
     // setForkEvery(1L)
+    environment(env.allVariables())
     retry {
         if (!buildId.contains("SNAPSHOT")) {
             maxRetries.set(3)

src/main/kotlin/com/katalisindonesia/banyuwangi/AppProperties.kt

@@ -40,4 +40,7 @@ data class AppProperties(
     val alarmHighMessages: Map<DetectionType, String>,
 
     val fcmRateLimit: Double,
+
+    val telegramToken: String,
+    val telegramRateLimit: Double = 30.0,
 )

src/main/kotlin/com/katalisindonesia/banyuwangi/consumer/TriggerConsumer.kt

@@ -5,6 +5,7 @@ import com.katalisindonesia.banyuwangi.model.SnapshotCount
 import com.katalisindonesia.banyuwangi.repo.AlarmRepo
 import com.katalisindonesia.banyuwangi.repo.SnapshotCountRepo
 import com.katalisindonesia.banyuwangi.service.AlarmService
+import com.katalisindonesia.banyuwangi.service.TelegramService
 import org.springframework.amqp.rabbit.annotation.RabbitListener
 import org.springframework.stereotype.Service
 import org.springframework.transaction.PlatformTransactionManager
@@ -16,6 +17,7 @@ class TriggerConsumer(
     private val alarmService: AlarmService,
     private val snapshotCountRepo: SnapshotCountRepo,
     transactionManager: PlatformTransactionManager,
+    private val telegramService: TelegramService,
 
 ) {
     private val tt = TransactionTemplate(transactionManager)
@@ -42,6 +44,7 @@ class TriggerConsumer(
                         alarm1
                     }!!
                     alarmService.sendAlarm(alarm)
+                    telegramService.sendAlarm(alarm)
                 }
             }
         }

src/main/kotlin/com/katalisindonesia/banyuwangi/model/TelegramChat.kt

@@ -0,0 +1,10 @@
+package com.katalisindonesia.banyuwangi.model
+
+import javax.persistence.Column
+import javax.persistence.Entity
+
+@Entity
+class TelegramChat(
+    @Column(unique = true)
+    val chatId: Long,
+) : Persistent()

src/main/kotlin/com/katalisindonesia/banyuwangi/repo/TelegramChatRepo.kt

@@ -0,0 +1,12 @@
+package com.katalisindonesia.banyuwangi.repo
+
+import com.katalisindonesia.banyuwangi.model.TelegramChat
+import org.springframework.stereotype.Repository
+import org.springframework.transaction.annotation.Transactional
+import java.util.UUID
+
+@Repository
+interface TelegramChatRepo : BaseRepository<TelegramChat, UUID> {
+    @Transactional
+    fun deleteByChatId(chatId: Long): Int
+}

src/main/kotlin/com/katalisindonesia/banyuwangi/service/AlarmService.kt

@@ -36,7 +36,7 @@ class AlarmService(
     fun sendAlarm(alarm: Alarm) {
         val type = alarm.snapshotCount.type
 
-        val titleBody = titleBody(alarm)
+        val titleBody = titleBody(alarm, appProperties)
         val title =
             MessageFormat.format(
                 titleBody.title ?: "", type.localizedName(), alarm.snapshotCount.snapshotCameraName
@@ -90,26 +90,9 @@ class AlarmService(
         rateLimit.acquire()
         firebaseMessaging.send(message)
     }
-
-    private fun titleBody(alarm: Alarm): TitleBody {
-        val type = alarm.snapshotCount.type
-        val value = alarm.snapshotCount.value
-        val minHighValue = appProperties.alarmHighMinimalValues[type]
-
-        if (minHighValue != null && value >= minHighValue) {
-            return TitleBody(
-                title = appProperties.alarmHighTitles[type] ?: appProperties.alarmTitles[type],
-                body = appProperties.alarmHighMessages[type] ?: appProperties.alarmMessages[type],
-            )
-        }
-        return TitleBody(
-            title = appProperties.alarmTitles[type],
-            body = appProperties.alarmMessages[type],
-        )
-    }
 }
 
-private data class TitleBody(
+data class TitleBody(
     val title: String?,
     val body: String?,
 )

src/main/kotlin/com/katalisindonesia/banyuwangi/service/Alarms.kt

@@ -0,0 +1,21 @@
+package com.katalisindonesia.banyuwangi.service
+
+import com.katalisindonesia.banyuwangi.AppProperties
+import com.katalisindonesia.banyuwangi.model.Alarm
+
+fun titleBody(alarm: Alarm, appProperties: AppProperties): TitleBody {
+    val type = alarm.snapshotCount.type
+    val value = alarm.snapshotCount.value
+    val minHighValue = appProperties.alarmHighMinimalValues[type]
+
+    if (minHighValue != null && value >= minHighValue) {
+        return TitleBody(
+            title = appProperties.alarmHighTitles[type] ?: appProperties.alarmTitles[type],
+            body = appProperties.alarmHighMessages[type] ?: appProperties.alarmMessages[type],
+        )
+    }
+    return TitleBody(
+        title = appProperties.alarmTitles[type],
+        body = appProperties.alarmMessages[type],
+    )
+}

src/main/kotlin/com/katalisindonesia/banyuwangi/service/TelegramService.kt

@@ -0,0 +1,107 @@
+package com.katalisindonesia.banyuwangi.service
+
+import com.github.kotlintelegrambot.bot
+import com.github.kotlintelegrambot.dispatch
+import com.github.kotlintelegrambot.dispatcher.command
+import com.github.kotlintelegrambot.dispatcher.telegramError
+import com.github.kotlintelegrambot.entities.ChatId
+import com.github.kotlintelegrambot.entities.TelegramFile
+import com.github.kotlintelegrambot.logging.LogLevel
+import com.google.common.util.concurrent.RateLimiter
+import com.katalisindonesia.banyuwangi.AppProperties
+import com.katalisindonesia.banyuwangi.model.Alarm
+import com.katalisindonesia.banyuwangi.model.TelegramChat
+import com.katalisindonesia.banyuwangi.repo.TelegramChatRepo
+import com.katalisindonesia.imageserver.service.StorageService
+import mu.KotlinLogging
+import org.springframework.dao.DataIntegrityViolationException
+import org.springframework.retry.support.RetryTemplate
+import org.springframework.stereotype.Service
+import java.text.MessageFormat
+import javax.annotation.PostConstruct
+
+private val log = KotlinLogging.logger { }
+
+@Service
+class TelegramService(
+    private val appProperties: AppProperties,
+    private val telegramChatRepo: TelegramChatRepo,
+    private val storageService: StorageService,
+) {
+    private val rateLimit = RateLimiter.create(appProperties.telegramRateLimit)
+
+    private val rt = RetryTemplate.builder().maxAttempts(10).exponentialBackoff(100L, 1.2, 10000L).build()
+
+    private val bot = bot {
+        token = appProperties.telegramToken
+        timeout = 30
+        logLevel = LogLevel.Network.Body
+
+        dispatch {
+            command("start") {
+                start(ChatId.fromId(update.message!!.chat.id))
+            }
+
+            command("stop") {
+                stop(ChatId.fromId(update.message!!.chat.id))
+            }
+
+            telegramError {
+                log.error { error.getErrorMessage() }
+            }
+        }
+    }
+
+    fun start(chatId: ChatId.Id) {
+        val chat = TelegramChat(chatId.id)
+        try {
+            telegramChatRepo.saveAndFlush(chat)
+            bot.sendMessage(
+                chatId = chatId,
+                text = "Anda telah berlangganan peringatan.\n\n" +
+                    "/stop untuk berhenti berlangganan"
+            )
+        } catch (e: DataIntegrityViolationException) {
+            log.debug(e) { "Duplicate subscription: " + chatId.id }
+            bot.sendMessage(chatId = chatId, text = "Anda sudah berlangganan")
+        }
+    }
+
+    fun stop(chatId: ChatId.Id) {
+        telegramChatRepo.deleteByChatId(chatId.id)
+        bot.sendMessage(
+            chatId = chatId,
+            text = "Anda telah berhenti berlangganan.\n\n" +
+                "/start untuk berlangganan peringatan"
+        )
+    }
+
+    @PostConstruct
+    fun init() {
+        bot.startPolling()
+    }
+
+    fun sendAlarm(alarm: Alarm) {
+        val type = alarm.snapshotCount.type
+
+        val titleBody = titleBody(alarm, appProperties)
+        val title =
+            MessageFormat.format(
+                titleBody.title ?: "", type.localizedName(), alarm.snapshotCount.snapshotCameraName
+            )
+        val body = titleBody.body ?: ""
+        val media = TelegramFile.ByFile(storageService.file(alarm.snapshotCount.snapshotImageId))
+
+        val chats = telegramChatRepo.findAll()
+        for (chat in chats) {
+            rateLimit.acquire()
+            rt.execute<Unit, Exception> {
+                bot.sendPhoto(
+                    ChatId.fromId(chat.chatId), media,
+                    caption = "$title\n\n$body\n\n" +
+                        "/stop untuk berhenti berlangganan"
+                )
+            }
+        }
+    }
+}

src/test/kotlin/com/katalisindonesia/banyuwangi/service/TelegramServiceTest.kt

@@ -0,0 +1,135 @@
+package com.katalisindonesia.banyuwangi.service
+
+import com.github.kotlintelegrambot.entities.ChatId
+import com.katalisindonesia.banyuwangi.model.Alarm
+import com.katalisindonesia.banyuwangi.model.Camera
+import com.katalisindonesia.banyuwangi.model.DetectionType
+import com.katalisindonesia.banyuwangi.model.Snapshot
+import com.katalisindonesia.banyuwangi.model.SnapshotCount
+import com.katalisindonesia.banyuwangi.repo.TelegramChatRepo
+import com.katalisindonesia.imageserver.service.StorageService
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.extension.ExtendWith
+import org.springframework.beans.factory.annotation.Autowired
+import org.springframework.boot.test.context.SpringBootTest
+import org.springframework.core.io.ClassPathResource
+import org.springframework.test.context.ActiveProfiles
+import org.springframework.test.context.junit.jupiter.SpringExtension
+import java.time.ZoneId
+import java.time.ZonedDateTime
+import java.util.UUID
+
+@SpringBootTest
+@ExtendWith(SpringExtension::class)
+@ActiveProfiles("default", "secret")
+class TelegramServiceTest(
+    @Autowired
+    private val telegramService: TelegramService,
+    @Autowired
+    private val telegramChatRepo: TelegramChatRepo,
+    @Autowired
+    private val storageService: StorageService,
+) {
+    @Test
+    fun test_start_stop() {
+        telegramService.start(ChatId.fromId(10))
+
+        assertEquals(1, telegramChatRepo.findAll().size)
+
+        // duplicate
+        telegramService.start(ChatId.fromId(10))
+        assertEquals(1, telegramChatRepo.findAll().size)
+
+        // different
+        telegramService.start(ChatId.fromId(59573981))
+        assertEquals(2, telegramChatRepo.findAll().size)
+
+        // stop
+        telegramService.stop(ChatId.fromId(59573981))
+        assertEquals(1, telegramChatRepo.findAll().size)
+
+        // duplicate stop
+        telegramService.stop(ChatId.fromId(59573981))
+        assertEquals(1, telegramChatRepo.findAll().size)
+
+        // all stop
+        telegramService.stop(ChatId.fromId(10))
+        assertEquals(0, telegramChatRepo.findAll().size)
+    }
+
+    @Test
+    fun sendAlarm() {
+        telegramService.start(ChatId.fromId(59573981))
+        val camera0 = Camera(name = "camera0", location = "")
+
+        val snapshot0 = Snapshot(imageId = imageId(), camera = camera0, length = 0, isAnnotation = true)
+
+        val base = ZonedDateTime.of(1970, 1, 1, 7, 0, 0, 0, ZoneId.systemDefault()).toInstant()
+        snapshot0.created = base.plusMillis(2100)
+
+        val count0 = SnapshotCount(
+            snapshot = snapshot0,
+            type = DetectionType.CROWD,
+            value = 2,
+        )
+        telegramService.sendAlarm(
+            Alarm(
+                maxValue = 1,
+                snapshotCount = count0,
+            )
+        )
+        telegramService.stop(ChatId.fromId(59573981))
+    }
+    @Test
+    fun sendAlarmFloodHigh() {
+        telegramService.start(ChatId.fromId(59573981))
+        val camera0 = Camera(name = "camera0", location = "")
+
+        val snapshot0 = Snapshot(imageId = imageId(), camera = camera0, length = 0, isAnnotation = true)
+
+        val base = ZonedDateTime.of(1970, 1, 1, 7, 0, 0, 0, ZoneId.systemDefault()).toInstant()
+        snapshot0.created = base.plusMillis(2100)
+
+        val count0 = SnapshotCount(
+            snapshot = snapshot0,
+            type = DetectionType.FLOOD,
+            value = 100,
+        )
+        telegramService.sendAlarm(
+            Alarm(
+                maxValue = 1,
+                snapshotCount = count0,
+            )
+        )
+        telegramService.stop(ChatId.fromId(59573981))
+    }
+    @Test
+    fun sendAlarmFloodLow() {
+        telegramService.start(ChatId.fromId(59573981))
+        val camera0 = Camera(name = "camera0", location = "")
+
+        val snapshot0 = Snapshot(imageId = imageId(), camera = camera0, length = 0, isAnnotation = true)
+
+        val base = ZonedDateTime.of(1970, 1, 1, 7, 0, 0, 0, ZoneId.systemDefault()).toInstant()
+        snapshot0.created = base.plusMillis(2100)
+
+        val count0 = SnapshotCount(
+            snapshot = snapshot0,
+            type = DetectionType.FLOOD,
+            value = 1,
+        )
+        telegramService.sendAlarm(
+            Alarm(
+                maxValue = 1,
+                snapshotCount = count0,
+            )
+        )
+        telegramService.stop(ChatId.fromId(59573981))
+    }
+
+    private fun imageId(): UUID {
+        val imageId = storageService.store(ClassPathResource("dog_bike_car.jpg").inputStream.readAllBytes())
+        return imageId
+    }
+}