feat: add SQS DeleteMessageBatch operation with performance optimizations

models/queue.go

@@ -48,6 +48,7 @@ type Queue interface {
 	Filter(tenantId int64, queue string, filterCriteria FilterCriteria) []int64
 
 	Delete(tenantId int64, queue string, messageId int64) error
+	DeleteBatch(tenantId int64, queue string, messageIds []int64) error
 
 	Shutdown() error
 }

protocols/sqs/models.go

@@ -145,10 +145,33 @@ type SendMessageBatchResultEntry struct {
 	SequenceNumber         string `json:"SequenceNumber,omitempty"`
 }
 
-// BatchResultErrorEntry represents a failed entry in the SendMessageBatch operation.
+// BatchResultErrorEntry represents a failed entry in batch operations.
 type BatchResultErrorEntry struct {
 	ID          string `json:"Id"`
 	SenderFault bool   `json:"SenderFault"`
 	Code        string `json:"Code"`
 	Message     string `json:"Message"`
 }
+
+// DeleteMessageBatchRequest represents the input for the DeleteMessageBatch operation.
+type DeleteMessageBatchRequest struct {
+	QueueUrl string                           `json:"QueueUrl"`
+	Entries  []DeleteMessageBatchRequestEntry `json:"Entries"`
+}
+
+// DeleteMessageBatchRequestEntry represents an entry in the DeleteMessageBatch operation.
+type DeleteMessageBatchRequestEntry struct {
+	ID            string `json:"Id"`
+	ReceiptHandle string `json:"ReceiptHandle"`
+}
+
+// DeleteMessageBatchResponse represents the output for the DeleteMessageBatch operation.
+type DeleteMessageBatchResponse struct {
+	Successful []DeleteMessageBatchResultEntry `json:"Successful"`
+	Failed     []BatchResultErrorEntry         `json:"Failed"`
+}
+
+// DeleteMessageBatchResultEntry represents a successful entry in the DeleteMessageBatch operation.
+type DeleteMessageBatchResultEntry struct {
+	ID string `json:"Id"`
+}

protocols/sqs/sqs.go

@@ -175,6 +175,8 @@ func (s *SQS) Action(c *fiber.Ctx) error {
 		rc = s.ReceiveMessage(c, tenantId)
 	case "AmazonSQS.DeleteMessage":
 		rc = s.DeleteMessage(c, tenantId)
+	case "AmazonSQS.DeleteMessageBatch":
+		rc = s.DeleteMessageBatch(c, tenantId)
 	case "AmazonSQS.ListQueues":
 		rc = s.ListQueues(c, tenantId)
 	case "AmazonSQS.GetQueueUrl":
@@ -596,6 +598,64 @@ func (s *SQS) DeleteMessage(c *fiber.Ctx, tenantId int64) error {
 	return nil
 }
 
+func (s *SQS) DeleteMessageBatch(c *fiber.Ctx, tenantId int64) error {
+	batchReq := &DeleteMessageBatchRequest{}
+
+	err := json.Unmarshal(c.Body(), batchReq)
+	if err != nil {
+		return err
+	}
+
+	tokens := strings.Split(batchReq.QueueUrl, "/")
+	queue := tokens[len(tokens)-1]
+
+	response := &DeleteMessageBatchResponse{}
+
+	// First pass: validate receipt handles and collect valid message IDs
+	validEntries := make([]DeleteMessageBatchRequestEntry, 0, len(batchReq.Entries))
+	messageIds := make([]int64, 0, len(batchReq.Entries))
+
+	for _, req := range batchReq.Entries {
+		messageId, err := strconv.ParseInt(req.ReceiptHandle, 10, 64)
+		if err != nil {
+			response.Failed = append(response.Failed, BatchResultErrorEntry{
+				ID:          req.ID,
+				SenderFault: true,
+				Code:        "InvalidParameterValue",
+				Message:     "Invalid ReceiptHandle",
+			})
+			continue
+		}
+		validEntries = append(validEntries, req)
+		messageIds = append(messageIds, messageId)
+	}
+
+	// Batch delete all valid messages at once
+	if len(messageIds) > 0 {
+		err = s.queue.DeleteBatch(tenantId, queue, messageIds)
+		if err != nil {
+			// If batch delete fails, mark all as failed
+			for _, req := range validEntries {
+				response.Failed = append(response.Failed, BatchResultErrorEntry{
+					ID:          req.ID,
+					SenderFault: false,
+					Code:        "InternalFailure",
+					Message:     err.Error(),
+				})
+			}
+		} else {
+			// All succeeded
+			for _, req := range validEntries {
+				response.Successful = append(response.Successful, DeleteMessageBatchResultEntry{
+					ID: req.ID,
+				})
+			}
+		}
+	}
+
+	return c.JSON(response)
+}
+
 func (s *SQS) ChangeMessageVisibility(c *fiber.Ctx, tenantId int64) error {
 	req := &ChangeMessageVisibilityRequest{}
 

queue/sqlite/sqlite.go

@@ -473,7 +473,8 @@ func (q *SQLiteQueue) Dequeue(tenantId int64, queueName string, numToDequeue int
 	}
 
 	err = q.DBG.Transaction(func(tx *gorm.DB) error {
-		res = tx.Model(&Message{}).Where("tenant_id = ? AND queue_id = ? AND id in ?", tenantId, queue.ID, messageIDs).
+		// Use primary key only for faster lookup - IDs already filtered by tenant/queue in Find above
+		res = tx.Model(&Message{}).Where("id IN ?", messageIDs).
 			UpdateColumns(map[string]any{
 				"tries":        gorm.Expr("tries+1"),
 				"delivered_at": now,
@@ -645,11 +646,13 @@ func (q *SQLiteQueue) Delete(tenantId int64, queueName string, messageId int64)
 	defer q.Mu.Unlock()
 
 	err = q.DBG.Transaction(func(tx *gorm.DB) error {
+		// KV uses composite index idx_kv(tenant_id, queue_id, message_id), so keep all filters
 		if err := tx.Where("tenant_id = ? AND queue_id = ? AND message_id = ?", tenantId, queue.ID, messageId).Delete(&KV{}).Error; err != nil {
 			return err
 		}
 
-		if err := tx.Where("tenant_id = ? AND queue_id = ? AND id = ?", tenantId, queue.ID, messageId).Delete(&Message{}).Error; err != nil {
+		// Message uses primary key on id, so query by id only for faster lookup
+		if err := tx.Where("id = ?", messageId).Delete(&Message{}).Error; err != nil {
 			return err
 		}
 
@@ -663,6 +666,40 @@ func (q *SQLiteQueue) Delete(tenantId int64, queueName string, messageId int64)
 	return err
 }
 
+func (q *SQLiteQueue) DeleteBatch(tenantId int64, queueName string, messageIds []int64) error {
+	if len(messageIds) == 0 {
+		return nil
+	}
+
+	queue, err := q.getQueue(tenantId, queueName)
+	if err != nil {
+		return err
+	}
+
+	q.Mu.Lock()
+	defer q.Mu.Unlock()
+
+	err = q.DBG.Transaction(func(tx *gorm.DB) error {
+		// KV uses composite index idx_kv(tenant_id, queue_id, message_id), so keep all filters
+		if err := tx.Where("tenant_id = ? AND queue_id = ? AND message_id IN ?", tenantId, queue.ID, messageIds).Delete(&KV{}).Error; err != nil {
+			return err
+		}
+
+		// Message uses primary key on id, so query by id only for faster lookup
+		if err := tx.Where("id IN ?", messageIds).Delete(&Message{}).Error; err != nil {
+			return err
+		}
+
+		return nil
+	})
+
+	if err == nil {
+		log.Debug().Interface("message_ids", messageIds).Msg("Deleted messages in batch")
+	}
+
+	return err
+}
+
 func (q *SQLiteQueue) Shutdown() error {
 	db, err := q.DBG.DB()
 	if err != nil {