Collapse Duplicate Loads with Singleflight

package main

import (
	"fmt"
	"sync"
	"sync/atomic"
	"time"
)

type call[T any] struct {
	wg  sync.WaitGroup
	val T
	err error
}

type Group[T any] struct {
	mu    sync.Mutex
	calls map[string]*call[T]
	loads atomic.Int64
}

func NewGroup[T any]() *Group[T] {
	return &Group[T]{calls: make(map[string]*call[T])}
}

func (g *Group[T]) Do(key string, load func() (T, error)) (T, bool, error) {
	g.mu.Lock()
	if existing, ok := g.calls[key]; ok {
		g.mu.Unlock()
		existing.wg.Wait()
		return existing.val, false, existing.err
	}

	current := &call[T]{}
	current.wg.Add(1)
	g.calls[key] = current
	g.loads.Add(1)
	g.mu.Unlock()

	current.val, current.err = load()
	current.wg.Done()

	g.mu.Lock()
	delete(g.calls, key)
	g.mu.Unlock()

	return current.val, true, current.err
}

func (g *Group[T]) Loads() int64 {
	return g.loads.Load()
}

func main() {
	group := NewGroup[string]()
	var wg sync.WaitGroup
	start := make(chan struct{})

	for id := 0; id < 8; id++ {
		wg.Add(1)
		go func(id int) {
			defer wg.Done()
			<-start
			value, leader, err := group.Do("user:42", func() (string, error) {
				time.Sleep(30 * time.Millisecond)
				return "profile-42", nil
			})
			fmt.Printf("caller %d leader=%t value=%s err=%v\n", id, leader, value, err)
		}(id)
	}

	close(start)
	wg.Wait()
	fmt.Println("loads:", group.Loads())
}