internal/webhooks/webhook_service_batch.go

Rama main · réplica de lectura del repositorio original.

package webhooks

import (
	"context"
	"fmt"
	"time"

	svcerrors "github.com/sarathsp06/sparrow/internal/errors"
	"github.com/sarathsp06/sparrow/internal/storage"
	"github.com/sarathsp06/sparrow/internal/tenant"
	"github.com/sarathsp06/sparrow/internal/webhooks/queue"
	"github.com/sarathsp06/sparrow/internal/webhooks/store"
)

// --- Batch Operations ---

// BatchScope says whose batch jobs a caller may reach. Consumer is the
// consumer in the request path. A job snapshotted for one consumer is
// reachable only under that consumer; a job snapshotted across all consumers
// (Consumer "") is reachable under any consumer path unless ConsumerOnly is
// set, which consumer-scoped callers (portal tokens) always set. A job outside
// the scope answers not found, so its existence is not revealed.
type BatchScope struct {
	Consumer     string
	ConsumerOnly bool
}

func (b BatchScope) allows(job *store.BatchJob) bool {
	if job.Consumer == b.Consumer {
		return true
	}
	return job.Consumer == "" && !b.ConsumerOnly
}

// loadAndValidateBatch parses a batch ID, loads the job, and verifies it is in
// scope and matches the expected job type. This eliminates the repeated
// parse+get+nil-check+type-check pattern across all batch operations.
func (s *WebhookService) loadAndValidateBatch(ctx context.Context, scope BatchScope, batchID string, expectedType store.BatchJobType) (*store.BatchJob, error) {
	batchUUID, err := parseUUID(batchID, "batch ID")
	if err != nil {
		return nil, err
	}

	batch, err := s.webhookRepo.GetBatchJob(ctx, tenant.FromContext(ctx), batchUUID)
	if err != nil {
		return nil, fmt.Errorf("failed to get batch job: %w", err)
	}
	if batch == nil || !scope.allows(batch) {
		return nil, svcerrors.Error(svcerrors.NotFound, "batch job not found")
	}
	if batch.JobType != expectedType {
		return nil, svcerrors.Errorf(svcerrors.FailedPrecondition, "batch job is not a %s", expectedType)
	}
	return batch, nil
}

// startBatch validates a pending batch job, transitions it to processing, and
// enqueues a River job for async execution. Used by RePushEvents and RetryDeliveries.
func (s *WebhookService) startBatch(ctx context.Context, scope BatchScope, batchID string, jobType store.BatchJobType) error {
	batch, err := s.loadAndValidateBatch(ctx, scope, batchID, jobType)
	if err != nil {
		return err
	}
	if batch.Status != store.BatchStatusPending {
		return svcerrors.Errorf(svcerrors.FailedPrecondition, "batch job is not in pending status (current: %s)", batch.Status)
	}
	if time.Now().After(batch.ExpiresAt) {
		return svcerrors.Error(svcerrors.FailedPrecondition, "batch job has expired")
	}

	// CAS pending→processing: only one caller wins; a concurrent start loses the race.
	if err := s.webhookRepo.UpdateBatchJobStatus(ctx, batch.TenantID, batch.ID, store.BatchStatusPending, store.BatchStatusProcessing); err != nil {
		if storage.IsNotFound(err) {
			return svcerrors.Error(svcerrors.FailedPrecondition, "batch job is no longer in pending status")
		}
		return fmt.Errorf("failed to update batch status: %w", err)
	}

	_, err = s.jobInserter.Insert(ctx, &queue.BatchJobArgs{
		TenantID: batch.TenantID.String(),
		BatchID:  batchID,
	})
	if err != nil {
		// Roll back status on enqueue failure
		_ = s.webhookRepo.UpdateBatchJobStatus(ctx, batch.TenantID, batch.ID, store.BatchStatusProcessing, store.BatchStatusPending)
		return fmt.Errorf("failed to enqueue batch job: %w", err)
	}

	s.logger.InfoContext(ctx, "Batch job enqueued", "batch_id", batchID, "job_type", jobType, "total", batch.Total)
	return nil
}

// cancelBatch validates a batch job and transitions it to cancelled.
// Used by CancelRepush and CancelRetry.
func (s *WebhookService) cancelBatch(ctx context.Context, scope BatchScope, batchID string, jobType store.BatchJobType) error {
	batch, err := s.loadAndValidateBatch(ctx, scope, batchID, jobType)
	if err != nil {
		return err
	}
	if batch.Status == store.BatchStatusCompleted || batch.Status == store.BatchStatusCancelled {
		return svcerrors.Errorf(svcerrors.FailedPrecondition, "batch job is already in terminal state: %s", batch.Status)
	}

	// CAS from the observed status: if it changed concurrently (e.g. completed), fail rather than clobber.
	if err := s.webhookRepo.UpdateBatchJobStatus(ctx, batch.TenantID, batch.ID, batch.Status, store.BatchStatusCancelled); err != nil {
		if storage.IsNotFound(err) {
			return svcerrors.Error(svcerrors.FailedPrecondition, "batch job status changed concurrently; retry cancel")
		}
		return fmt.Errorf("failed to cancel batch job: %w", err)
	}

	s.logger.InfoContext(ctx, "Batch job cancelled", "batch_id", batchID, "job_type", jobType)
	return nil
}

// RePushEvents starts async processing of a batch re-push.
// The batch must exist, belong to the tenant, be of type event_repush, and be in pending status.
func (s *WebhookService) RePushEvents(ctx context.Context, scope BatchScope, repushID string) error {
	return s.startBatch(ctx, scope, repushID, store.BatchTypeEventRepush)
}

// GetRepushStatus returns the current state of a batch re-push.
func (s *WebhookService) GetRepushStatus(ctx context.Context, scope BatchScope, repushID string) (*store.BatchJob, error) {
	return s.loadAndValidateBatch(ctx, scope, repushID, store.BatchTypeEventRepush)
}

// CancelRepush aborts a pending or in-progress batch re-push.
func (s *WebhookService) CancelRepush(ctx context.Context, scope BatchScope, repushID string) error {
	return s.cancelBatch(ctx, scope, repushID, store.BatchTypeEventRepush)
}

// RetryDeliveries starts async processing of a batch delivery retry.
func (s *WebhookService) RetryDeliveries(ctx context.Context, scope BatchScope, retryID string) error {
	return s.startBatch(ctx, scope, retryID, store.BatchTypeDeliveryRetry)
}

// GetRetryStatus returns the current state of a batch delivery retry.
func (s *WebhookService) GetRetryStatus(ctx context.Context, scope BatchScope, retryID string) (*store.BatchJob, error) {
	return s.loadAndValidateBatch(ctx, scope, retryID, store.BatchTypeDeliveryRetry)
}

// CancelRetry aborts a pending or in-progress batch delivery retry.
func (s *WebhookService) CancelRetry(ctx context.Context, scope BatchScope, retryID string) error {
	return s.cancelBatch(ctx, scope, retryID, store.BatchTypeDeliveryRetry)
}