feat(bot): add multi-household setup flow

apps/bot/src/purchase-topic-ingestion.ts

@@ -2,6 +2,10 @@ import { parsePurchaseMessage, type PurchaseParserLlmFallback } from '@household
 import { and, eq } from 'drizzle-orm'
 import type { Bot, Context } from 'grammy'
 import type { Logger } from '@household/observability'
+import type {
+  HouseholdConfigurationRepository,
+  HouseholdTopicBindingRecord
+} from '@household/ports'
 
 import { createDbClient, schema } from '@household/db'
 
@@ -61,6 +65,30 @@ export function extractPurchaseTopicCandidate(
   }
 }
 
+export function resolveConfiguredPurchaseTopicRecord(
+  value: PurchaseTopicCandidate,
+  binding: HouseholdTopicBindingRecord
+): PurchaseTopicRecord | null {
+  if (value.rawText.trim().startsWith('/')) {
+    return null
+  }
+
+  if (binding.role !== 'purchase') {
+    return null
+  }
+
+  const normalizedText = value.rawText.trim()
+  if (normalizedText.length === 0) {
+    return null
+  }
+
+  return {
+    ...value,
+    rawText: normalizedText,
+    householdId: binding.householdId
+  }
+}
+
 function needsReviewAsInt(value: boolean): number {
   return value ? 1 : 0
 }
@@ -245,3 +273,69 @@ export function registerPurchaseTopicIngestion(
     }
   })
 }
+
+export function registerConfiguredPurchaseTopicIngestion(
+  bot: Bot,
+  householdConfigurationRepository: HouseholdConfigurationRepository,
+  repository: PurchaseMessageIngestionRepository,
+  options: {
+    llmFallback?: PurchaseParserLlmFallback
+    logger?: Logger
+  } = {}
+): void {
+  bot.on('message:text', async (ctx, next) => {
+    const candidate = toCandidateFromContext(ctx)
+    if (!candidate) {
+      await next()
+      return
+    }
+
+    const binding = await householdConfigurationRepository.findHouseholdTopicByTelegramContext({
+      telegramChatId: candidate.chatId,
+      telegramThreadId: candidate.threadId
+    })
+
+    if (!binding) {
+      await next()
+      return
+    }
+
+    const record = resolveConfiguredPurchaseTopicRecord(candidate, binding)
+    if (!record) {
+      await next()
+      return
+    }
+
+    try {
+      const status = await repository.save(record, options.llmFallback)
+
+      if (status === 'created') {
+        options.logger?.info(
+          {
+            event: 'purchase.ingested',
+            householdId: record.householdId,
+            chatId: record.chatId,
+            threadId: record.threadId,
+            messageId: record.messageId,
+            updateId: record.updateId,
+            senderTelegramUserId: record.senderTelegramUserId
+          },
+          'Purchase topic message ingested'
+        )
+      }
+    } catch (error) {
+      options.logger?.error(
+        {
+          event: 'purchase.ingest_failed',
+          householdId: record.householdId,
+          chatId: record.chatId,
+          threadId: record.threadId,
+          messageId: record.messageId,
+          updateId: record.updateId,
+          error
+        },
+        'Failed to ingest purchase topic message'
+      )
+    }
+  })
+}