go-kata/01-context-cancellation-concurrency/01-concurrent-aggregator/aggregator.go

package concurrent_aggregator

import (
	"context"
	"fmt"
	"log/slog"
	"time"

	"github.com/medunes/go-kata/01-context-cancellation-concurrency/01-concurrent-aggregator/order"
	"github.com/medunes/go-kata/01-context-cancellation-concurrency/01-concurrent-aggregator/profile"

	"golang.org/x/sync/errgroup"
)

type Option func(ua *UserAggregator)

func WithLogger(l *slog.Logger) Option {
	return func(ua *UserAggregator) {
		ua.logger = l
	}
}

func WithTimeout(timeout time.Duration) Option {
	return func(ua *UserAggregator) {
		ua.timeout = timeout
	}
}

type UserAggregator struct {
	orderService   order.Service
	profileService profile.Service
	timeout        time.Duration
	logger         *slog.Logger
}

func NewUserAggregator(os order.Service, ps profile.Service, opts ...Option) *UserAggregator {
	ua := &UserAggregator{
		orderService:   os,
		profileService: ps,
		logger:         slog.Default(), // avoid nil panics, default writes to stderr
	}
	for _, opt := range opts {
		opt(ua)
	}
	return ua
}

type AggregatedProfile struct {
	Name string
	Cost float64
}

func (ua *UserAggregator) Aggregate(ctx context.Context, id int) ([]*AggregatedProfile, error) {
	var (
		localCtx context.Context
		cancel   context.CancelFunc
		pr       *profile.Profile
		or       []*order.Order
		au       []*AggregatedProfile
		g        *errgroup.Group
	)
	if ua.timeout > 0 {
		localCtx, cancel = context.WithTimeout(ctx, ua.timeout)
	} else {
		localCtx, cancel = context.WithCancel(ctx)
	}
	defer cancel()
	ua.logger.Info("starting aggregation", "user_id", id)
	g, localCtx = errgroup.WithContext(localCtx)
	g.Go(func() error {
		var err error
		pr, err = ua.profileService.Get(localCtx, id)
		if err != nil {
			ua.logger.Warn("failed to fetch profile", "user_id", id, "err", err)
			return fmt.Errorf("profile fetch failed: %w", err)
		}
		return nil
	})
	g.Go(func() error {
		var err error
		or, err = ua.orderService.GetAll(localCtx, id)
		if err != nil {
			ua.logger.Warn("failed to fetch order", "user_id", id, "err", err)
			return fmt.Errorf("order fetch failed: %w", err)
		}
		return nil

	})

	err := g.Wait()
	if err != nil {
		ua.logger.Warn("aggregator exited with error", "user_id", id, "err", err)
		return nil, err
	}
	for _, o := range or {
		if o.UserId == id {
			au = append(au, &AggregatedProfile{pr.Name, o.Cost})
		}
	}
	ua.logger.Info("aggregation complete successfully", "user_id", id, "count", len(au))
	return au, nil
}