monobudget import: WIP

src/main/kotlin/io/github/smaugfm/monobudget/Application.kt

@@ -4,8 +4,8 @@ import io.github.oshai.kotlinlogging.KotlinLogging
 import io.github.smaugfm.monobudget.common.BaseApplication
 import io.github.smaugfm.monobudget.common.startup.ApplicationStartupVerifier
 import io.github.smaugfm.monobudget.common.statement.StatementSource
-import io.github.smaugfm.monobudget.common.telegram.TelegramApi
-import io.github.smaugfm.monobudget.common.telegram.TelegramCallbackHandler
+import io.github.smaugfm.monobudget.common.notify.TelegramApi
+import io.github.smaugfm.monobudget.common.notify.TelegramCallbackHandler
 import io.github.smaugfm.monobudget.common.util.injectAll
 import org.koin.core.component.inject
 import kotlin.system.exitProcess

src/main/kotlin/io/github/smaugfm/monobudget/common/account/TransferCache.kt

@@ -5,5 +5,22 @@ import io.github.smaugfm.monobudget.common.util.misc.ConcurrentExpiringMap
 import kotlinx.coroutines.Deferred
 import kotlin.time.Duration
 
-abstract class TransferCache<TTransaction>(expirationDuration: Duration) :
-    ConcurrentExpiringMap<StatementItem, Deferred<TTransaction>>(expirationDuration)
+interface TransferCache<TTransaction> {
+    suspend fun getEntries(item: StatementItem): Set<Map.Entry<StatementItem, Deferred<TTransaction>>>
+
+    suspend fun put(item: StatementItem, transaction: Deferred<TTransaction>)
+
+    open class Expiring<TTransaction>(expirationDuration: Duration) :
+        TransferCache<TTransaction> {
+        private val cache = ConcurrentExpiringMap<StatementItem, Deferred<TTransaction>>(expirationDuration)
+
+        override suspend fun getEntries(item: StatementItem):
+            Set<Map.Entry<StatementItem, Deferred<TTransaction>>> {
+            return cache.entries
+        }
+
+        override suspend fun put(item: StatementItem, transaction: Deferred<TTransaction>) {
+            cache.add(item, transaction)
+        }
+    }
+}

src/main/kotlin/io/github/smaugfm/monobudget/common/account/TransferDetector.kt

@@ -3,21 +3,19 @@ package io.github.smaugfm.monobudget.common.account
 import io.github.oshai.kotlinlogging.KotlinLogging
 import io.github.smaugfm.monobudget.common.model.financial.StatementItem
 import io.github.smaugfm.monobudget.common.statement.lifecycle.StatementProcessingContext
-import io.github.smaugfm.monobudget.common.util.misc.ConcurrentExpiringMap
 import kotlinx.coroutines.CompletableDeferred
-import kotlinx.coroutines.Deferred
 
 private val log = KotlinLogging.logger {}
 
 abstract class TransferDetector<TTransaction>(
     private val bankAccounts: BankAccountService,
     private val ctx: StatementProcessingContext,
-    private val cache: ConcurrentExpiringMap<StatementItem, Deferred<TTransaction>>,
+    private val cache: TransferCache<TTransaction>,
 ) {
     suspend fun checkForTransfer(): MaybeTransfer<TTransaction> =
         ctx.getOrPut("transfer") {
             val existingTransfer =
-                cache.entries.firstOrNull { (recentStatementItem) ->
+                cache.getEntries(ctx.item).firstOrNull { (recentStatementItem) ->
                     checkIsTransferTransactions(recentStatementItem)
                 }?.value?.await()
 
@@ -30,7 +28,7 @@ abstract class TransferDetector<TTransaction>(
                 MaybeTransfer.Transfer(ctx.item, existingTransfer)
             } else {
                 val deferred = CompletableDeferred<TTransaction>()
-                cache.add(ctx.item, deferred)
+                cache.put(ctx.item, deferred)
 
                 MaybeTransfer.NotTransfer(ctx.item, deferred)
             }

src/main/kotlin/io/github/smaugfm/monobudget/common/notify/StatementItemNotificationSender.kt

@@ -0,0 +1,8 @@
+package io.github.smaugfm.monobudget.common.notify
+
+import io.github.smaugfm.monobudget.common.model.financial.BankAccountId
+import io.github.smaugfm.monobudget.common.model.telegram.MessageWithReplyKeyboard
+
+interface StatementItemNotificationSender {
+    suspend fun notify(accountId: BankAccountId, newMessage: MessageWithReplyKeyboard)
+}

src/main/kotlin/io/github/smaugfm/monobudget/common/telegram/TelegramApi.kt → src/main/kotlin/io/github/smaugfm/monobudget/common/notify/TelegramApi.kt

@@ -1,4 +1,4 @@
-package io.github.smaugfm.monobudget.common.telegram
+package io.github.smaugfm.monobudget.common.notify
 
 import com.elbekd.bot.Bot
 import com.elbekd.bot.model.ChatId

src/main/kotlin/io/github/smaugfm/monobudget/common/telegram/TelegramCallbackHandler.kt → src/main/kotlin/io/github/smaugfm/monobudget/common/notify/TelegramCallbackHandler.kt

@@ -1,4 +1,4 @@
-package io.github.smaugfm.monobudget.common.telegram
+package io.github.smaugfm.monobudget.common.notify
 
 import com.elbekd.bot.model.ChatId
 import com.elbekd.bot.model.TelegramApiError

src/main/kotlin/io/github/smaugfm/monobudget/common/telegram/TelegramErrorHandlerEventListener.kt → src/main/kotlin/io/github/smaugfm/monobudget/common/notify/TelegramErrorHandlerEventListener.kt

@@ -1,4 +1,4 @@
-package io.github.smaugfm.monobudget.common.telegram
+package io.github.smaugfm.monobudget.common.notify
 
 import com.elbekd.bot.model.ChatId
 import com.elbekd.bot.types.CallbackQuery

src/main/kotlin/io/github/smaugfm/monobudget/common/telegram/TelegramMessageSender.kt → src/main/kotlin/io/github/smaugfm/monobudget/common/notify/TelegramNotificationSender.kt

@@ -1,4 +1,4 @@
-package io.github.smaugfm.monobudget.common.telegram
+package io.github.smaugfm.monobudget.common.notify
 
 import com.elbekd.bot.model.ChatId
 import com.elbekd.bot.types.ParseMode
@@ -11,11 +11,11 @@ import org.koin.core.annotation.Single
 private val log = KotlinLogging.logger {}
 
 @Single
-class TelegramMessageSender(
+class TelegramNotificationSender(
     private val bankAccounts: BankAccountService,
     private val telegramApi: TelegramApi,
-) {
-    suspend fun send(
+): StatementItemNotificationSender {
+    override suspend fun notify(
         accountId: BankAccountId,
         newMessage: MessageWithReplyKeyboard,
     ) {

src/main/kotlin/io/github/smaugfm/monobudget/common/statement/lifecycle/StatementItemProcessor.kt

@@ -3,7 +3,7 @@ package io.github.smaugfm.monobudget.common.statement.lifecycle
 import io.github.oshai.kotlinlogging.KotlinLogging
 import io.github.smaugfm.monobudget.common.account.BankAccountService
 import io.github.smaugfm.monobudget.common.account.TransferDetector
-import io.github.smaugfm.monobudget.common.telegram.TelegramMessageSender
+import io.github.smaugfm.monobudget.common.notify.StatementItemNotificationSender
 import io.github.smaugfm.monobudget.common.transaction.TransactionFactory
 import io.github.smaugfm.monobudget.common.transaction.TransactionMessageFormatter
 import io.github.smaugfm.monobudget.common.util.pp
@@ -16,7 +16,7 @@ abstract class StatementItemProcessor<TTransaction, TNewTransaction>(
     private val bankAccounts: BankAccountService,
     private val transferDetector: TransferDetector<TTransaction>,
     private val messageFormatter: TransactionMessageFormatter<TTransaction>,
-    private val telegramMessageSender: TelegramMessageSender,
+    private val notificationSender: StatementItemNotificationSender,
 ) {
     suspend fun process() {
         logStatement()
@@ -30,7 +30,7 @@ abstract class StatementItemProcessor<TTransaction, TNewTransaction>(
         val transaction = transactionFactory.create(maybeTransfer)
         val message = messageFormatter.format(ctx.item, transaction)
 
-        telegramMessageSender.send(ctx.item.accountId, message)
+        notificationSender.notify(ctx.item.accountId, message)
     }
 
     private suspend fun logStatement() {

src/main/kotlin/io/github/smaugfm/monobudget/importer/ImportApplication.kt

@@ -4,10 +4,12 @@ import io.github.oshai.kotlinlogging.KotlinLogging
 import io.github.smaugfm.lunchmoney.model.LunchmoneyInsertTransaction
 import io.github.smaugfm.lunchmoney.model.LunchmoneyTransaction
 import io.github.smaugfm.monobudget.common.BaseApplication
+import io.github.smaugfm.monobudget.common.account.TransferCache
 import io.github.smaugfm.monobudget.common.model.settings.Settings
+import io.github.smaugfm.monobudget.common.notify.StatementItemNotificationSender
 import io.github.smaugfm.monobudget.common.retry.InMemoryStatementRetryRepository
+import io.github.smaugfm.monobudget.common.transaction.NewTransactionFactory
 import io.github.smaugfm.monobudget.lunchmoney.LunchmoneyNewTransactionFactory
-import io.github.smaugfm.monobudget.lunchmoney.LunchmoneyTransferCache
 import io.github.smaugfm.monobudget.mono.MonoWebhookSettings
 import io.github.smaugfm.monobudget.setupKoinModules
 import kotlinx.coroutines.CoroutineScope
@@ -42,8 +44,13 @@ class ImportApplication(source: ImportStatementSource) :
                     MonoWebhookSettings(false, URI.create("none://none"), 0)
                 )
                 modules(module {
-                    single { LunchmoneyNewTransactionFactory(noteSuffix) }
-                    single { LunchmoneyTransferCache(Long.MAX_VALUE.seconds) }
+                    single<NewTransactionFactory<LunchmoneyInsertTransaction>> {
+                        LunchmoneyNewTransactionFactory(noteSuffix)
+                    }
+                    single<TransferCache<LunchmoneyTransaction>> {
+                        ImportTransferCache(7.seconds)
+                    }
+                    single<StatementItemNotificationSender> { ImportNotificationSender }
                 })
             }.koin
 

src/main/kotlin/io/github/smaugfm/monobudget/importer/ImportNotificationSender.kt

@@ -0,0 +1,11 @@
+package io.github.smaugfm.monobudget.importer
+
+import io.github.smaugfm.monobudget.common.model.financial.BankAccountId
+import io.github.smaugfm.monobudget.common.model.telegram.MessageWithReplyKeyboard
+import io.github.smaugfm.monobudget.common.notify.StatementItemNotificationSender
+
+object ImportNotificationSender : StatementItemNotificationSender {
+    override suspend fun notify(accountId: BankAccountId, newMessage: MessageWithReplyKeyboard) {
+        // no-op
+    }
+}

src/main/kotlin/io/github/smaugfm/monobudget/importer/ImportStatementSource.kt

@@ -39,7 +39,7 @@ class ImportStatementSource(private val configs: List<ImportAccountConfig>) :
             )
 
             log.info { "Loaded ${csvItems.size} transactions for ${config.accountAlias}" }
-            csvItems.map { it.toStatementItem(config.accountAlias, accountCurrency) }
+            csvItems.map { it.toStatementItem(accountId, accountCurrency) }
         }.flatten().sortedBy { it.time }
     }
 

src/main/kotlin/io/github/smaugfm/monobudget/importer/ImportTransferCache.kt

@@ -0,0 +1,36 @@
+package io.github.smaugfm.monobudget.importer
+
+import io.github.smaugfm.lunchmoney.model.LunchmoneyTransaction
+import io.github.smaugfm.monobudget.common.account.TransferCache
+import io.github.smaugfm.monobudget.common.model.financial.StatementItem
+import kotlinx.coroutines.Deferred
+import kotlinx.coroutines.DelicateCoroutinesApi
+import kotlinx.coroutines.newSingleThreadContext
+import kotlinx.coroutines.withContext
+import kotlin.time.Duration
+
+@OptIn(DelicateCoroutinesApi::class)
+class ImportTransferCache(private val cacheTimeSimulatedDuration: Duration) :
+    TransferCache<LunchmoneyTransaction> {
+    private val singleThreadedContext = newSingleThreadContext("import-transfer-cache")
+    private val cache = mutableMapOf<StatementItem, Deferred<LunchmoneyTransaction>>()
+        .toSortedMap { o1, o2 -> o1.time.compareTo(o2.time) }
+
+    @Suppress("DeferredResultUnused")
+    override suspend fun getEntries(item: StatementItem): Set<Map.Entry<StatementItem, Deferred<LunchmoneyTransaction>>> =
+        withContext(singleThreadedContext) {
+            val threshold = item.time - cacheTimeSimulatedDuration
+            while (cache.isNotEmpty()) {
+                val cur = cache.firstKey()
+                if (cur.time < threshold) {
+                    cache.remove(cur)
+                }
+            }
+            cache.entries
+        }
+
+    override suspend fun put(item: StatementItem, transaction: Deferred<LunchmoneyTransaction>) {
+        check(cache.keys.all { item.time >= it.time })
+        cache[item] = transaction
+    }
+}

src/main/kotlin/io/github/smaugfm/monobudget/lunchmoney/LunchmoneyNewTransactionFactory.kt

@@ -13,7 +13,7 @@ import java.util.Currency
 private val log = KotlinLogging.logger {}
 
 @Single
-class LunchmoneyNewTransactionFactory(val noteSuffix: String = "") :
+class LunchmoneyNewTransactionFactory(private val noteSuffix: String = "") :
     NewTransactionFactory<LunchmoneyInsertTransaction>() {
 
     override suspend fun create(statement: StatementItem): LunchmoneyInsertTransaction {

src/main/kotlin/io/github/smaugfm/monobudget/lunchmoney/LunchmoneyStatementItemProcessor.kt

@@ -4,10 +4,10 @@ import io.github.smaugfm.lunchmoney.model.LunchmoneyInsertTransaction
 import io.github.smaugfm.lunchmoney.model.LunchmoneyTransaction
 import io.github.smaugfm.monobudget.common.account.BankAccountService
 import io.github.smaugfm.monobudget.common.account.TransferDetector
+import io.github.smaugfm.monobudget.common.notify.StatementItemNotificationSender
 import io.github.smaugfm.monobudget.common.statement.lifecycle.StatementItemProcessor
 import io.github.smaugfm.monobudget.common.statement.lifecycle.StatementProcessingContext
 import io.github.smaugfm.monobudget.common.statement.lifecycle.StatementProcessingScopeComponent
-import io.github.smaugfm.monobudget.common.telegram.TelegramMessageSender
 import io.github.smaugfm.monobudget.common.transaction.TransactionFactory
 import io.github.smaugfm.monobudget.common.transaction.TransactionMessageFormatter
 import org.koin.core.annotation.Scope
@@ -21,12 +21,12 @@ class LunchmoneyStatementItemProcessor(
     bankAccounts: BankAccountService,
     transferDetector: TransferDetector<LunchmoneyTransaction>,
     messageFormatter: TransactionMessageFormatter<LunchmoneyTransaction>,
-    telegramMessageSender: TelegramMessageSender,
+    notificationSender: StatementItemNotificationSender,
 ) : StatementItemProcessor<LunchmoneyTransaction, LunchmoneyInsertTransaction>(
         ctx,
         transactionFactory,
         bankAccounts,
         transferDetector,
         messageFormatter,
-        telegramMessageSender,
+        notificationSender,
     )

src/main/kotlin/io/github/smaugfm/monobudget/lunchmoney/LunchmoneyTelegramCallbackHandler.kt

@@ -8,7 +8,7 @@ import io.github.smaugfm.lunchmoney.model.LunchmoneyUpdateTransaction
 import io.github.smaugfm.lunchmoney.model.enumeration.LunchmoneyTransactionStatus
 import io.github.smaugfm.monobudget.common.category.CategoryService
 import io.github.smaugfm.monobudget.common.model.callback.TransactionUpdateType
-import io.github.smaugfm.monobudget.common.telegram.TelegramCallbackHandler
+import io.github.smaugfm.monobudget.common.notify.TelegramCallbackHandler
 import io.github.smaugfm.monobudget.common.transaction.TransactionMessageFormatter.Companion.extractFromOldMessage
 import io.github.smaugfm.monobudget.common.transaction.TransactionMessageFormatter.Companion.formatHTMLStatementMessage
 import io.github.smaugfm.monobudget.lunchmoney.LunchmoneyTransactionMessageFormatter.Companion.constructTransactionsQuickUrl

src/main/kotlin/io/github/smaugfm/monobudget/lunchmoney/LunchmoneyTransferCache.kt

@@ -8,4 +8,4 @@ import kotlin.time.Duration.Companion.minutes
 
 @Single
 class LunchmoneyTransferCache(expirationDuration: Duration = 1.minutes) :
-    TransferCache<LunchmoneyTransaction>(expirationDuration)
+    TransferCache.Expiring<LunchmoneyTransaction>(expirationDuration)

src/main/kotlin/io/github/smaugfm/monobudget/lunchmoney/LunchmoneyTransferDetector.kt

@@ -2,6 +2,7 @@ package io.github.smaugfm.monobudget.lunchmoney
 
 import io.github.smaugfm.lunchmoney.model.LunchmoneyTransaction
 import io.github.smaugfm.monobudget.common.account.BankAccountService
+import io.github.smaugfm.monobudget.common.account.TransferCache
 import io.github.smaugfm.monobudget.common.account.TransferDetector
 import io.github.smaugfm.monobudget.common.statement.lifecycle.StatementProcessingContext
 import io.github.smaugfm.monobudget.common.statement.lifecycle.StatementProcessingScopeComponent
@@ -13,5 +14,5 @@ import org.koin.core.annotation.Scoped
 class LunchmoneyTransferDetector(
     bankAccounts: BankAccountService,
     ctx: StatementProcessingContext,
-    cache: LunchmoneyTransferCache,
+    cache: TransferCache<LunchmoneyTransaction>,
 ) : TransferDetector<LunchmoneyTransaction>(bankAccounts, ctx, cache)

src/main/kotlin/io/github/smaugfm/monobudget/ynab/YnabStatementItemProcessor.kt

@@ -2,10 +2,10 @@ package io.github.smaugfm.monobudget.ynab
 
 import io.github.smaugfm.monobudget.common.account.BankAccountService
 import io.github.smaugfm.monobudget.common.account.TransferDetector
+import io.github.smaugfm.monobudget.common.notify.StatementItemNotificationSender
 import io.github.smaugfm.monobudget.common.statement.lifecycle.StatementItemProcessor
 import io.github.smaugfm.monobudget.common.statement.lifecycle.StatementProcessingContext
 import io.github.smaugfm.monobudget.common.statement.lifecycle.StatementProcessingScopeComponent
-import io.github.smaugfm.monobudget.common.telegram.TelegramMessageSender
 import io.github.smaugfm.monobudget.common.transaction.TransactionFactory
 import io.github.smaugfm.monobudget.common.transaction.TransactionMessageFormatter
 import io.github.smaugfm.monobudget.ynab.model.YnabSaveTransaction
@@ -21,12 +21,12 @@ class YnabStatementItemProcessor(
     bankAccounts: BankAccountService,
     transferDetector: TransferDetector<YnabTransactionDetail>,
     messageFormatter: TransactionMessageFormatter<YnabTransactionDetail>,
-    telegramMessageSender: TelegramMessageSender,
+    notificationSender: StatementItemNotificationSender,
 ) : StatementItemProcessor<YnabTransactionDetail, YnabSaveTransaction>(
         ctx,
         transactionFactory,
         bankAccounts,
         transferDetector,
         messageFormatter,
-        telegramMessageSender,
+        notificationSender,
     )

src/main/kotlin/io/github/smaugfm/monobudget/ynab/YnabTelegramCallbackHandler.kt

@@ -2,7 +2,7 @@ package io.github.smaugfm.monobudget.ynab
 
 import com.elbekd.bot.types.Message
 import io.github.smaugfm.monobudget.common.model.callback.TransactionUpdateType
-import io.github.smaugfm.monobudget.common.telegram.TelegramCallbackHandler
+import io.github.smaugfm.monobudget.common.notify.TelegramCallbackHandler
 import io.github.smaugfm.monobudget.common.transaction.TransactionMessageFormatter.Companion.extractFromOldMessage
 import io.github.smaugfm.monobudget.common.transaction.TransactionMessageFormatter.Companion.formatHTMLStatementMessage
 import io.github.smaugfm.monobudget.ynab.model.YnabTransactionDetail

src/main/kotlin/io/github/smaugfm/monobudget/ynab/YnabTransferCache.kt

@@ -8,4 +8,4 @@ import kotlin.time.Duration.Companion.minutes
 
 @Single
 class YnabTransferCache(expirationDuration: Duration = 1.minutes) :
-    TransferCache<YnabTransactionDetail>(expirationDuration)
+    TransferCache.Expiring<YnabTransactionDetail>(expirationDuration)

src/main/kotlin/io/github/smaugfm/monobudget/ynab/YnabTransferDetector.kt

@@ -1,6 +1,7 @@
 package io.github.smaugfm.monobudget.ynab
 
 import io.github.smaugfm.monobudget.common.account.BankAccountService
+import io.github.smaugfm.monobudget.common.account.TransferCache
 import io.github.smaugfm.monobudget.common.account.TransferDetector
 import io.github.smaugfm.monobudget.common.statement.lifecycle.StatementProcessingContext
 import io.github.smaugfm.monobudget.common.statement.lifecycle.StatementProcessingScopeComponent
@@ -13,7 +14,7 @@ import org.koin.core.annotation.Scoped
 class YnabTransferDetector(
     bankAccounts: BankAccountService,
     ctx: StatementProcessingContext,
-    cache: YnabTransferCache,
+    cache: TransferCache<YnabTransactionDetail>,
 ) : TransferDetector<YnabTransactionDetail>(
         bankAccounts,
         ctx,

src/test/kotlin/io/github/smaugfm/monobudget/integration/util/IntegrationTestBase.kt

@@ -16,7 +16,7 @@ import io.github.smaugfm.monobudget.common.startup.ApplicationStartupVerifier
 import io.github.smaugfm.monobudget.common.startup.BudgetSettingsVerifier
 import io.github.smaugfm.monobudget.common.statement.StatementSource
 import io.github.smaugfm.monobudget.common.statement.lifecycle.StatementProcessingContext
-import io.github.smaugfm.monobudget.common.telegram.TelegramApi
+import io.github.smaugfm.monobudget.common.notify.TelegramApi
 import io.github.smaugfm.monobudget.common.util.misc.PeriodicFetcherFactory
 import io.github.smaugfm.monobudget.integration.TransactionsTest
 import io.github.smaugfm.monobudget.mono.MonoWebhookListener