Merge pull request #1943 from qwibitai/fix/pending-rows-idempotent

fix(delivery): make pending_questions/approvals insert idempotent

src/db/sessions.ts

@@ -97,10 +97,16 @@ export function deleteSession(id: string): void {
 
 // ── Pending Questions ──
 
-export function createPendingQuestion(pq: PendingQuestion): void {
-  getDb()
+/**
+ * Insert a pending question row. Idempotent: when delivery fails and retries,
+ * the second attempt calls this with the same question_id — without `OR
+ * IGNORE` that would throw UNIQUE and prevent the retry from reaching the
+ * actual send step. Returns true if a new row was inserted.
+ */
+export function createPendingQuestion(pq: PendingQuestion): boolean {
+  const result = getDb()
     .prepare(
-      `INSERT INTO pending_questions (question_id, session_id, message_out_id, platform_id, channel_type, thread_id, title, options_json, created_at)
+      `INSERT OR IGNORE INTO pending_questions (question_id, session_id, message_out_id, platform_id, channel_type, thread_id, title, options_json, created_at)
        VALUES (@question_id, @session_id, @message_out_id, @platform_id, @channel_type, @thread_id, @title, @options_json, @created_at)`,
     )
     .run({
@@ -114,6 +120,7 @@ export function createPendingQuestion(pq: PendingQuestion): void {
       options_json: JSON.stringify(pq.options),
       created_at: pq.created_at,
     });
+  return result.changes > 0;
 }
 
 export function getPendingQuestion(questionId: string): PendingQuestion | undefined {
@@ -131,16 +138,21 @@ export function deletePendingQuestion(questionId: string): void {
 
 // ── Pending Approvals ──
 
+/**
+ * Insert a pending approval row. Idempotent for the same reason as
+ * createPendingQuestion: delivery retries with the same approval_id must not
+ * fail on UNIQUE before the send step gets a chance to succeed.
+ */
 export function createPendingApproval(
   pa: Partial<PendingApproval> &
     Pick<
       PendingApproval,
       'approval_id' | 'request_id' | 'action' | 'payload' | 'created_at' | 'title' | 'options_json'
     >,
-): void {
-  getDb()
+): boolean {
+  const result = getDb()
     .prepare(
-      `INSERT INTO pending_approvals
+      `INSERT OR IGNORE INTO pending_approvals
          (approval_id, session_id, request_id, action, payload, created_at,
           agent_group_id, channel_type, platform_id, platform_message_id, expires_at, status,
           title, options_json)
@@ -159,6 +171,7 @@ export function createPendingApproval(
       status: 'pending',
       ...pa,
     });
+  return result.changes > 0;
 }
 
 export function getPendingApproval(approvalId: string): PendingApproval | undefined {

src/delivery.ts

@@ -321,7 +321,7 @@ async function deliverMessage(
         questionId: content.questionId,
       });
     } else {
-      createPendingQuestion({
+      const inserted = createPendingQuestion({
         question_id: content.questionId,
         session_id: session.id,
         message_out_id: msg.id,
@@ -332,7 +332,9 @@ async function deliverMessage(
         options: normalizeOptions(rawOptions as never),
         created_at: new Date().toISOString(),
       });
-      log.info('Pending question created', { questionId: content.questionId, sessionId: session.id });
+      if (inserted) {
+        log.info('Pending question created', { questionId: content.questionId, sessionId: session.id });
+      }
     }
   }