Async Pipelines

The pipeline package provides Go-native concurrent processing utilities. Instead of transplanting JavaScript Promises, it embraces goroutines, channels, and sync.WaitGroup to deliver clean, type-safe async patterns that feel at home in Go.

Concept

Every function in this package accepts a typed slice and a typed callback, uses Go generics for full type safety, and handles goroutine lifecycle, error collection, and synchronization for you. You focus on the business logic — the pipeline handles the concurrency.

import "github.com/CodeSyncr/nimbus/pipeline"

// Process 100 images, max 5 at a time
errs := pipeline.Pool(images, 5, func(img Image) error {
    return processAndUpload(img)
})

Available Functions

Sequential

Runs jobs one at a time. Stops immediately on the first error — perfect for ordered operations where later steps depend on earlier ones.

steps := []func() error{validateInput, createUser, sendWelcomeEmail}

err := pipeline.Sequential(steps, func(step func() error) error {
    return step()
})

Parallel & ParallelMap

Parallel fires all items concurrently. ParallelMap does the same but maps each item to a result, preserving the original order.

// Fire-and-forget notifications
errs := pipeline.Parallel(userIDs, func(id uint) error {
    return notification.Send(id, "Your order shipped!")
})

// Fetch prices from multiple APIs concurrently
prices, errs := pipeline.ParallelMap(symbols, func(sym string) (float64, error) {
    return fetchStockPrice(sym)
})
// prices[0] corresponds to symbols[0], guaranteed

Pool (Worker Pool)

Limits concurrency using a semaphore pattern. Crucial for I/O-bound tasks where you'd overwhelm an external service with unlimited goroutines.

// Resize images, max 10 concurrent goroutines
errs := pipeline.Pool(images, 10, func(img Image) error {
    resized, err := resize(img, 800, 600)
    if err != nil {
        return err
    }
    return storage.Put(img.Key, resized)
})

Retry

Retries a function up to N times with a configurable delay between attempts. Ideal for flaky network calls.

err := pipeline.Retry(3, 2*time.Second, func() error {
    return paymentGateway.Charge(order.Total)
})
if err != nil {
    // All 3 attempts failed
    return c.JSON(500, map[string]string{"error": "payment failed after retries"})
}

WithTimeout

Wraps any function with a deadline. If the function doesn't complete in time, it returns context.DeadlineExceeded.

result, err := pipeline.WithTimeout(5*time.Second, func() (APIResponse, error) {
    return externalAPI.FetchData(query)
})
if errors.Is(err, context.DeadlineExceeded) {
    return c.JSON(504, map[string]string{"error": "upstream timeout"})
}

Real-Life Example: E-Commerce Order Fulfillment

func fulfillOrder(order Order) error {
    // Step 1: Sequential — these MUST happen in order
    err := pipeline.Sequential([]string{"validate", "charge", "reserve"}, func(step string) error {
        switch step {
        case "validate":
            return validateInventory(order)
        case "charge":
            return pipeline.Retry(3, time.Second, func() error {
                return chargePayment(order)
            })
        case "reserve":
            return reserveStock(order)
        }
        return nil
    })
    if err != nil {
        return err
    }

    // Step 2: Parallel — these can happen simultaneously
    pipeline.Parallel(order.Items, func(item OrderItem) error {
        return generateShippingLabel(item)
    })

    return nil
}

Real-Life Example: Concurrent Data Import with Pool

func importCSVRows(rows []CSVRow) {
    // Insert into DB with max 20 concurrent connections
    errs := pipeline.Pool(rows, 20, func(row CSVRow) error {
        product := Product{
            Name:  row.Name,
            Price: row.Price,
            SKU:   row.SKU,
        }
        return database.Get().Create(&product).Error
    })

    for _, err := range errs {
        logger.Error("import failed", "error", err)
    }
}

When to Use What

• Sequential — Ordered steps (validation → payment → fulfillment)
• Parallel — Fire-and-forget (notifications, cache warming)
• Pool — I/O-bound tasks with external limits (DB writes, API calls, file uploads)
• ParallelMap — Fetch/transform data concurrently and collect results
• Retry — Flaky external services (payments, webhooks, third-party APIs)
• WithTimeout — Protect against hanging upstream calls