package com.octaldata.session.pulsar

import com.octaldata.domain.DeploymentConnectionConfig
import com.octaldata.domain.topics.TopicName
import com.octaldata.session.SubscriptionOps
import kotlinx.coroutines.future.await
import org.apache.pulsar.client.admin.PulsarAdmin
import kotlin.time.Duration

class PulsarSubscriptionOps(private val config: DeploymentConnectionConfig.Pulsar) : SubscriptionOps {

   private val client = PulsarAdmin.builder()
      .serviceHttpUrl(config.adminServiceUrl)
      .build()

   override suspend fun getSubscriptions(topicName: TopicName): Result<List<String>> = runCatching {
      client.topics().getSubscriptionsAsync(topicName.value).await()
   }

   override suspend fun deleteSubscription(topicName: TopicName, subscription: String): Result<Boolean> = runCatching {
      client.topics().deleteSubscriptionAsync(topicName.value, subscription).await()
      true
   }

   override suspend fun expireSub(topicName: TopicName, time: Duration): Result<Boolean> = runCatching {
      client.topics().expireMessagesForAllSubscriptionsAsync(topicName.value, time.inWholeSeconds).await()
      true
   }

   override suspend fun expireMessages(topicName: TopicName, subscription: String, time: Duration): Result<Boolean> =
      runCatching {
         client.topics().expireMessagesAsync(topicName.value, subscription, time.inWholeSeconds).await()
         true
      }

   override suspend fun skipAllMessages(topicName: TopicName, subscription: String): Result<Boolean> = runCatching {
      client.topics().skipAllMessagesAsync(topicName.value, subscription).await()
      true
   }
}
