internal/webhooks/webhook_service_delivery.go
Rama main · réplica de lectura del repositorio original.
package webhooks
import (
"context"
"database/sql"
"fmt"
"github.com/google/uuid"
"github.com/riverqueue/river"
svcerrors "github.com/sarathsp06/sparrow/internal/errors"
"github.com/sarathsp06/sparrow/internal/tenant"
"github.com/sarathsp06/sparrow/internal/webhooks/queue"
"github.com/sarathsp06/sparrow/internal/webhooks/store"
)
// GetDeliveryStatus gets the status of a webhook delivery.
// When consumer is empty, looks up by delivery ID alone.
func (s *WebhookService) GetDeliveryStatus(ctx context.Context, deliveryID string, consumer string) (*store.WebhookDelivery, error) {
s.logger.InfoContext(ctx, "Getting webhook delivery status",
"delivery_id", deliveryID,
"consumer", consumer)
if deliveryID == "" {
return nil, svcerrors.Error(svcerrors.InvalidArgument, "delivery ID is required")
}
tenantID := tenant.FromContext(ctx)
id, err := parseUUID(deliveryID, "delivery ID")
if err != nil {
return nil, err
}
delivery, err := s.webhookRepo.GetDeliveryByID(ctx, tenantID, id, consumer)
if err != nil {
s.logger.ErrorContext(ctx, "Failed to get delivery by ID", "error", err)
return nil, fmt.Errorf("failed to retrieve delivery status: %w", err)
}
if delivery == nil {
return nil, svcerrors.Error(svcerrors.NotFound, "delivery not found")
}
return delivery, nil
}
// GetDeliveryAttempts retrieves individual attempt history for a delivery.
// Returns all recorded health events for the delivery, ordered by timestamp ascending.
// When consumer is set, a delivery of another consumer answers not found;
// when empty, it looks up by delivery ID alone.
func (s *WebhookService) GetDeliveryAttempts(ctx context.Context, deliveryID string, consumer string) ([]*store.WebhookHealthEvent, error) {
tenantID := tenant.FromContext(ctx)
s.logger.InfoContext(ctx, "Getting delivery attempts", "delivery_id", deliveryID, "tenant_id", tenantID.String())
if deliveryID == "" {
return nil, svcerrors.Error(svcerrors.InvalidArgument, "delivery ID is required")
}
id, err := parseUUID(deliveryID, "delivery ID")
if err != nil {
return nil, err
}
if consumer != "" {
delivery, err := s.webhookRepo.GetDeliveryByID(ctx, tenantID, id, consumer)
if err != nil {
return nil, fmt.Errorf("failed to retrieve delivery: %w", err)
}
if delivery == nil {
return nil, svcerrors.Error(svcerrors.NotFound, "delivery not found")
}
}
attempts, err := s.webhookRepo.GetDeliveryAttempts(ctx, tenantID, id)
if err != nil {
s.logger.ErrorContext(ctx, "Failed to get delivery attempts", "error", err)
return nil, fmt.Errorf("failed to retrieve delivery attempts: %w", err)
}
return attempts, nil
}
// ListDeliveries retrieves delivery history with filters.
// Supports filtering by consumer, webhook, event, status, error_category,
// subscription, and time range via the DeliveryFilter struct.
// When PrepareRetry is true, snapshots all matching delivery IDs into a batch job and returns the batch ID.
func (s *WebhookService) ListDeliveries(ctx context.Context, filter store.DeliveryFilter) (*DeliveryPage, error) {
s.logger.InfoContext(ctx, "Listing deliveries",
"consumer", filter.Consumer,
"webhook_id", filter.WebhookID,
"event_id", filter.EventID,
"status", filter.Status,
"error_category", filter.ErrorCategory,
"prepare_retry", filter.PrepareRetry,
"limit", filter.Limit)
tenantID := tenant.FromContext(ctx)
filter.Limit, _ = normalizePagination(filter.Limit, 0)
deliveries, hasMore, err := s.webhookRepo.ListDeliveriesFiltered(ctx, tenantID, filter)
if err != nil {
s.logger.ErrorContext(ctx, "Failed to list deliveries", "error", err)
return nil, fmt.Errorf("failed to retrieve deliveries: %w", err)
}
// Snapshot matching IDs into a batch job if requested
page := &DeliveryPage{Items: deliveries, HasMore: hasMore}
if filter.PrepareRetry {
ids, err := s.webhookRepo.SnapshotDeliveryIDs(ctx, tenantID, filter)
if err != nil {
s.logger.ErrorContext(ctx, "Failed to snapshot delivery IDs for retry", "error", err)
return nil, fmt.Errorf("failed to prepare retry: %w", err)
}
if len(ids) > 0 {
filterMap := map[string]any{
"consumer": filter.Consumer,
}
if filter.WebhookID != nil {
filterMap["webhook_id"] = filter.WebhookID.String()
}
if filter.EventID != nil {
filterMap["event_id"] = filter.EventID.String()
}
if filter.Status != nil {
filterMap["status"] = *filter.Status
}
if filter.ErrorCategory != nil {
filterMap["error_category"] = *filter.ErrorCategory
}
if filter.EventName != nil {
filterMap["event_name"] = *filter.EventName
}
batchData := &store.BatchJobData{
ItemIDs: ids,
Filter: filterMap,
}
batchJob, err := s.webhookRepo.CreateBatchJob(ctx, tenantID, filter.Consumer, store.BatchTypeDeliveryRetry, batchData)
if err != nil {
s.logger.ErrorContext(ctx, "Failed to create batch job for retry", "error", err)
return nil, fmt.Errorf("failed to create retry batch: %w", err)
}
page.RetryID = batchJob.ID.String()
page.RetryTotal = len(ids)
s.logger.InfoContext(ctx, "Created retry batch job",
"retry_id", page.RetryID,
"delivery_count", len(ids))
}
}
return page, nil
}
// retryChunkSize bounds each reset UPDATE and job BatchInsert of a retry.
const retryChunkSize = 1000
// RetryDelivery manually retries failed or pending webhook deliveries
func (s *WebhookService) RetryDelivery(ctx context.Context, consumer string, deliveryID string, webhookID string, force bool) ([]string, int32, error) {
s.logger.InfoContext(ctx, "Processing retry delivery request",
"delivery_id", deliveryID,
"webhook_id", webhookID,
"consumer", consumer,
"force", force)
tenantID := tenant.FromContext(ctx)
// Validate required fields
if deliveryID == "" && webhookID == "" {
return nil, 0, svcerrors.Error(svcerrors.InvalidArgument, "either delivery_id or webhook_id is required")
}
// Consumer is required for webhook-level retry (multiple deliveries),
// but optional for single-delivery retry (delivery_id is globally unique within a tenant).
if consumer == "" && webhookID != "" {
return nil, 0, svcerrors.Error(svcerrors.InvalidArgument, "consumer is required for webhook-level retry")
}
if deliveryID != "" && webhookID != "" {
return nil, 0, svcerrors.Error(svcerrors.InvalidArgument, "only one of delivery_id or webhook_id can be specified")
}
var deliveriesToResubmit []*store.WebhookDelivery
// Every delivery retried here belongs to this one webhook: a single
// delivery's, or the webhook named in the request.
var webhook *store.WebhookRegistration
if deliveryID != "" {
id, err := parseUUID(deliveryID, "delivery ID")
if err != nil {
return nil, 0, err
}
// Resubmit specific delivery
delivery, err := s.webhookRepo.GetDeliveryByID(ctx, tenantID, id, consumer)
if err != nil {
s.logger.ErrorContext(ctx, "Failed to get delivery", "error", err)
return nil, 0, fmt.Errorf("failed to retrieve delivery: %w", err)
}
if delivery == nil {
return nil, 0, svcerrors.Error(svcerrors.NotFound, "delivery not found")
}
// Check if delivery can be resubmitted
if !force && delivery.Status == store.StatusSuccess {
return nil, 0, svcerrors.Error(svcerrors.FailedPrecondition, "delivery already succeeded. Use force to resubmit anyway")
}
deliveriesToResubmit = []*store.WebhookDelivery{delivery}
webhook, err = s.webhookRepo.GetWebhookByID(ctx, tenantID, delivery.WebhookID, consumer)
if err != nil {
s.logger.ErrorContext(ctx, "Failed to get webhook for delivery", "webhook_id", delivery.WebhookID, "error", err)
return nil, 0, fmt.Errorf("failed to retrieve webhook: %w", err)
}
} else {
id, err := parseUUID(webhookID, "webhook ID")
if err != nil {
return nil, 0, err
}
// Resubmit all failed/pending deliveries for webhook
webhook, err = s.webhookRepo.GetWebhookByID(ctx, tenantID, id, consumer)
if err != nil {
s.logger.ErrorContext(ctx, "Failed to get webhook", "error", err)
return nil, 0, fmt.Errorf("failed to retrieve webhook: %w", err)
}
// Get retriable deliveries
deliveriesToResubmit, err = s.webhookRepo.GetRetriableDeliveries(ctx, tenantID, id, consumer, force)
if err != nil {
s.logger.ErrorContext(ctx, "Failed to get retriable deliveries", "error", err)
return nil, 0, fmt.Errorf("failed to retrieve deliveries: %w", err)
}
if len(deliveriesToResubmit) == 0 {
message := "No failed or pending deliveries found"
if !force {
message += ". Use force to resubmit all deliveries"
}
s.logger.InfoContext(ctx, message)
return []string{}, 0, nil
}
}
// One UPDATE for every delivery and one insert for every job, rather
// than a round trip (and a webhook lookup) per delivery.
ids := make([]uuid.UUID, len(deliveriesToResubmit))
jobs := make(map[uuid.UUID]river.JobArgs, len(deliveriesToResubmit))
for i, delivery := range deliveriesToResubmit {
ids[i] = delivery.ID
// Manual retries never expire -- use far-future sentinel so TTL doesn't apply.
args := &queue.WebhookArgs{
DeliveryID: delivery.ID.String(),
WebhookID: delivery.WebhookID.String(),
EventID: delivery.EventID.String(),
ExpiresAt: store.NoExpiryTime,
Consumer: webhook.Consumer,
TenantID: tenantID.String(),
MaxAttempts: delivery.MaxAttempts,
RetryBackoffSeconds: webhook.RetryBackoffSeconds,
}
if delivery.SubscriptionID != nil {
args.SubscriptionID = delivery.SubscriptionID.String()
}
jobs[delivery.ID] = args
}
// In chunks, so a forced retry of a large webhook stays a bounded
// statement and insert. Each chunk resets its finished deliveries, and
// retrying ones whose waiting job it cancels, and queues their jobs in
// one transaction; ones being sent right now are left to their job.
resubmittedIDs := make([]string, 0, len(ids))
for start := 0; start < len(ids); start += retryChunkSize {
end := min(start+retryChunkSize, len(ids))
reset, err := s.webhookRepo.ResetDeliveriesForRetry(ctx, tenantID, ids[start:end], s.jobInserter.CancelWaitingDeliveryJobsTx, func(ctx context.Context, tx *sql.Tx, reset []uuid.UUID) error {
chunk := make([]river.JobArgs, len(reset))
for i, id := range reset {
chunk[i] = jobs[id]
}
return s.jobInserter.BatchInsertTx(ctx, tx, chunk)
})
if err != nil {
s.logger.ErrorContext(ctx, "Failed to resubmit deliveries", "count", end-start, "error", err)
return nil, 0, fmt.Errorf("failed to resubmit deliveries: %w", err)
}
for _, id := range reset {
resubmittedIDs = append(resubmittedIDs, id.String())
}
}
resubmittedCount := clampInt32(len(resubmittedIDs))
if resubmittedCount == 0 {
return nil, 0, svcerrors.Error(svcerrors.FailedPrecondition, "nothing to resubmit: the delivery is being sent right now or is queued to send; retry it again once that attempt finishes")
}
s.logger.InfoContext(ctx, "Webhook deliveries resubmitted successfully",
"resubmitted_count", resubmittedCount,
"total_requested", len(deliveriesToResubmit))
return resubmittedIDs, resubmittedCount, nil
}