package workercluster

import (
	"context"
	"fmt"
	"sync"
	"time"
)

// BootstrapResult is the outcome of a Bootstrap call.
type BootstrapResult struct {
	// Leader is the node id that ended up holding leadership when
	// bootstrap finished. For a 1-node bootstrap this is always the
	// local node.
	Leader string

	// Voters is the final voter set as observed on the local node.
	Voters []string
}

// Bootstrap starts a fresh cluster from the first node. It is a
// convenience wrapper around New + Start + BootstrapCluster for the
// common single-node bootstrap case. Production code that needs
// custom timeouts should call New / Start directly.
func Bootstrap(ctx context.Context, opts *Options) (*Cluster, BootstrapResult, error) {
	opts.Bootstrap = true
	opts.Seed = Peer{}
	c, err := New(opts)
	if err != nil {
		return nil, BootstrapResult{}, err
	}
	if err := c.Start(ctx); err != nil {
		return nil, BootstrapResult{}, err
	}

	if err := waitForLeader(ctx, c, 10*time.Second); err != nil {
		_ = c.Shutdown(ctx)
		return nil, BootstrapResult{}, err
	}

	stats := c.Stats()
	return c, BootstrapResult{Leader: stats.Leader, Voters: votersFromStats(&stats)}, nil
}

// JoinCluster brings up a new node that joins an existing cluster via
// the given seed peer. It blocks until the local node is a voter in
// the raft configuration.
func JoinCluster(ctx context.Context, opts *Options, seed Peer) (*Cluster, error) {
	opts.Bootstrap = false
	opts.Seed = seed
	c, err := New(opts)
	if err != nil {
		return nil, err
	}
	if err := c.Start(ctx); err != nil {
		return nil, err
	}
	if err := waitForLeader(ctx, c, 15*time.Second); err != nil {
		_ = c.Shutdown(ctx)
		return nil, err
	}
	return c, nil
}

// waitForLeader blocks until the cluster has a leader (or ctx is
// canceled, or timeout elapses).
func waitForLeader(ctx context.Context, c *Cluster, timeout time.Duration) error {
	deadline := time.Now().Add(timeout)
	for {
		stats := c.Stats()
		if stats.Leader != "" {
			return nil
		}
		if time.Now().After(deadline) {
			return fmt.Errorf("workercluster: no leader after %s (state=%s)", timeout, stats.State)
		}
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-time.After(50 * time.Millisecond):
		}
	}
}

// votersFromStats returns the voter WorkerIDs from the cluster stats.
// It is best-effort: the FSM membership cache is the source of truth.
func votersFromStats(s *Stats) []string {
	out := make([]string, 0, len(s.Members))
	for _, m := range s.Members {
		if m.Role == RoleVoter {
			out = append(out, m.WorkerID)
		}
	}
	return out
}

// Parallel starts the given cluster starts concurrently and returns
// once all of them have completed (or the first one errors).
func Parallel(starts ...func() error) error {
	var wg sync.WaitGroup
	errs := make(chan error, len(starts))
	for _, fn := range starts {
		wg.Add(1)
		go func(fn func() error) {
			defer wg.Done()
			if err := fn(); err != nil {
				errs <- err
			}
		}(fn)
	}
	wg.Wait()
	close(errs)
	for err := range errs {
		if err != nil {
			return err
		}
	}
	return nil
}