// Package manager owns the running set of sync jobs.
//
// Each repository gets its own goroutine with its own timer, so a slow or
// broken repo never delays the others. A shared semaphore caps how many git
// processes run at once, which is what actually bounds memory use.
//
// Reloading is incremental: Apply diffs the new configuration against the
// running jobs and touches only what changed. Repos whose settings are
// unchanged keep their timer and their in-flight work.
package manager

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"os"
	"path/filepath"
	"reflect"
	"sort"
	"sync"
	"time"

	"syncbot/internal/config"
	"syncbot/internal/gitx"
	"syncbot/internal/syncer"
)

// Status is the observable state of one repository.
type Status struct {
	Name        string    `json:"name"`
	Src         string    `json:"src"`
	Dst         string    `json:"dst"`
	Interval    string    `json:"interval"`
	Syncing     bool      `json:"syncing"`
	Refs        int       `json:"refs"`
	Syncs       int64     `json:"syncs"`
	Pushes      int64     `json:"pushes"`
	Failures    int       `json:"consecutive_failures"`
	LastRun     time.Time `json:"last_run,omitzero"`
	LastSuccess time.Time `json:"last_success,omitzero"`
	NextRun     time.Time `json:"next_run,omitzero"`
	LastError   string    `json:"last_error,omitempty"`
	LastDurMS   int64     `json:"last_duration_ms"`
}

// Manager supervises one goroutine per repository.
type Manager struct {
	ctx context.Context
	log *slog.Logger

	mu      sync.Mutex
	jobs    map[string]*job
	stats   map[string]*Status
	sem     chan struct{}
	conc    int
	workDir string
	sync    *syncer.Syncer
	started time.Time

	// wg covers every goroutine ever started, including ones already replaced
	// by a reload, so shutdown does not leave a git process behind.
	wg sync.WaitGroup
}

type job struct {
	spec   config.Job
	cancel context.CancelFunc
	done   chan struct{}
}

// New returns a Manager whose jobs all derive from ctx.
func New(ctx context.Context, log *slog.Logger) *Manager {
	return &Manager{
		ctx:     ctx,
		log:     log,
		jobs:    make(map[string]*job),
		stats:   make(map[string]*Status),
		sem:     make(chan struct{}, 1),
		conc:    1,
		started: time.Now(),
	}
}

// Apply reconciles the running jobs with cfg. It never blocks on in-flight
// syncs: a job being replaced is cancelled, and its successor waits for the
// handover before touching the same mirror directory.
func (m *Manager) Apply(cfg *config.Config) error {
	home, err := prepareDirs(cfg.WorkDir)
	if err != nil {
		return err
	}

	if cfg.Concurrency <= 0 {
		cfg.Concurrency = 1
	}

	m.mu.Lock()

	if m.conc != cfg.Concurrency {
		// Callers release into the channel they acquired from, so swapping the
		// semaphore is safe even while syncs are in flight.
		m.log.Info("concurrency changed", "from", m.conc, "to", cfg.Concurrency)
		m.sem = make(chan struct{}, cfg.Concurrency)
		m.conc = cfg.Concurrency
	}

	workDirChanged := m.workDir != cfg.WorkDir
	if workDirChanged {
		m.workDir = cfg.WorkDir
		m.sync = &syncer.Syncer{Log: m.log, Home: home}
	}

	want := make(map[string]config.Job, len(cfg.Jobs))
	for _, j := range cfg.Jobs {
		want[j.Name] = j
	}

	// Stop jobs that disappeared or whose resolved spec changed.
	var stopped []*job
	for name, j := range m.jobs {
		w, keep := want[name]
		if keep && !workDirChanged && reflect.DeepEqual(w, j.spec) {
			continue
		}
		delete(m.jobs, name)
		stopped = append(stopped, j)
		if !keep {
			delete(m.stats, name)
			m.log.Info("repo removed", "repo", name)
		}
	}
	handover := make(map[string]<-chan struct{}, len(stopped))
	for _, j := range stopped {
		j.cancel()
		handover[j.spec.Name] = j.done
	}

	// Start whatever is now missing.
	var added, restarted int
	for i, spec := range cfg.Jobs {
		if _, running := m.jobs[spec.Name]; running {
			continue
		}
		prev := handover[spec.Name]
		if prev != nil {
			restarted++
		} else {
			added++
		}
		m.startLocked(spec, prev, stagger(i))
	}
	m.mu.Unlock()

	if added+restarted+len(stopped) > 0 {
		m.log.Info("configuration applied",
			"repos", len(cfg.Jobs), "started", added, "restarted", restarted,
			"stopped", len(stopped)-restarted, "concurrency", cfg.Concurrency)
	}
	return nil
}

// startLocked launches a job goroutine. m.mu must be held.
func (m *Manager) startLocked(spec config.Job, waitFor <-chan struct{}, delay time.Duration) {
	ctx, cancel := context.WithCancel(m.ctx)
	j := &job{spec: spec, cancel: cancel, done: make(chan struct{})}
	m.jobs[spec.Name] = j

	st, ok := m.stats[spec.Name]
	if !ok {
		st = &Status{Name: spec.Name}
		m.stats[spec.Name] = st
	}
	// Counters survive a restart; the descriptive fields follow the new spec.
	st.Src = gitx.RedactURL(spec.Src.URL)
	st.Dst = gitx.RedactURL(spec.Dst.URL)
	st.Interval = spec.Interval.String()

	m.wg.Add(1)
	go m.loop(ctx, j, waitFor, delay)
}

func (m *Manager) loop(ctx context.Context, j *job, waitFor <-chan struct{}, delay time.Duration) {
	defer m.wg.Done()
	defer close(j.done)

	// When replacing a job, let the previous goroutine finish unwinding before
	// operating on its mirror directory.
	if waitFor != nil {
		select {
		case <-waitFor:
		case <-ctx.Done():
			return
		}
	}

	log := m.log.With("repo", j.spec.Name)
	log.Info("watching", "src", gitx.RedactURL(j.spec.Src.URL),
		"dst", gitx.RedactURL(j.spec.Dst.URL), "interval", j.spec.Interval)

	timer := time.NewTimer(delay)
	defer timer.Stop()

	for {
		m.update(j.spec.Name, func(s *Status) { s.NextRun = time.Now().Add(delay) })

		select {
		case <-ctx.Done():
			return
		case <-timer.C:
		}

		failures := m.runOnce(ctx, j.spec, log)
		if ctx.Err() != nil {
			return
		}

		delay = j.spec.Interval
		if failures > 0 {
			delay = backoff(j.spec.Interval, j.spec.MaxBackoff, failures)
			if delay > j.spec.Interval {
				log.Warn("backing off", "failures", failures, "retry_in", delay)
			}
		}
		timer.Reset(delay)
	}
}

// runOnce performs a single sync and returns the consecutive failure count.
func (m *Manager) runOnce(ctx context.Context, spec config.Job, log *slog.Logger) int {
	release, err := m.acquire(ctx)
	if err != nil {
		return 0 // shutting down
	}
	defer release()

	m.update(spec.Name, func(s *Status) {
		s.Syncing = true
		s.LastRun = time.Now()
		s.Syncs++
	})

	res, err := m.syncer().Sync(ctx, spec)

	failures := 0
	m.update(spec.Name, func(s *Status) {
		s.Syncing = false
		s.LastDurMS = res.Duration.Milliseconds()
		if err != nil {
			s.Failures++
			s.LastError = err.Error()
		} else {
			s.Failures = 0
			s.LastError = ""
			s.LastSuccess = time.Now()
			s.Refs = res.Refs
			if res.Pushed {
				s.Pushes++
			}
		}
		failures = s.Failures
	})

	switch {
	case err != nil && ctx.Err() != nil:
		log.Info("sync cancelled")
	case err != nil:
		log.Error("sync failed", "err", err, "consecutive_failures", failures)
	case res.Pushed:
		log.Info("pushed", "refs", res.Refs, "changes", len(res.Changes),
			"dur", res.Duration.Round(time.Millisecond))
		for _, c := range res.Changes {
			log.Debug("ref updated", "change", c)
		}
	}
	return failures
}

// RunOnce syncs every job exactly once and reports whether all succeeded. It
// is the engine behind the -once flag, for cron-style or one-shot deployments.
func (m *Manager) RunOnce(cfg *config.Config) error {
	if err := m.Apply(&config.Config{ // reuse Apply's dir/semaphore setup, no jobs
		WorkDir: cfg.WorkDir, Concurrency: cfg.Concurrency,
		ReloadInterval: cfg.ReloadInterval, LogLevel: cfg.LogLevel, LogFormat: cfg.LogFormat,
	}); err != nil {
		return err
	}

	var wg sync.WaitGroup
	errs := make([]error, len(cfg.Jobs))
	for i, spec := range cfg.Jobs {
		wg.Add(1)
		go func() {
			defer wg.Done()
			release, err := m.acquire(m.ctx)
			if err != nil {
				errs[i] = err
				return
			}
			defer release()

			log := m.log.With("repo", spec.Name)
			res, err := m.syncer().Sync(m.ctx, spec)
			if err != nil {
				errs[i] = fmt.Errorf("%s: %w", spec.Name, err)
				log.Error("sync failed", "err", err)
				return
			}
			log.Info("synced", "refs", res.Refs, "pushed", res.Pushed,
				"dur", res.Duration.Round(time.Millisecond))
		}()
	}
	wg.Wait()
	return errors.Join(errs...)
}

// Stop cancels every job and waits up to grace for them to unwind. No further
// Apply may be called afterwards.
func (m *Manager) Stop(grace time.Duration) {
	m.mu.Lock()
	for _, j := range m.jobs {
		j.cancel()
	}
	m.jobs = make(map[string]*job)
	m.mu.Unlock()

	// Waiting on the group rather than on the current jobs also covers
	// goroutines a recent reload replaced but that have not finished unwinding.
	done := make(chan struct{})
	go func() {
		m.wg.Wait()
		close(done)
	}()

	select {
	case <-done:
	case <-time.After(grace):
		m.log.Warn("shutdown grace period expired; abandoning in-flight syncs", "grace", grace)
	}
}

// Statuses returns a snapshot of every known repository, sorted by name.
func (m *Manager) Statuses() []Status {
	m.mu.Lock()
	defer m.mu.Unlock()
	out := make([]Status, 0, len(m.stats))
	for _, s := range m.stats {
		out = append(out, *s)
	}
	sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name })
	return out
}

// Uptime reports how long the manager has been running.
func (m *Manager) Uptime() time.Duration { return time.Since(m.started) }

// acquire takes a slot from the concurrency semaphore. The release function
// returns the slot to the same channel it came from, which keeps the count
// correct even if Apply swapped the semaphore in the meantime.
func (m *Manager) acquire(ctx context.Context) (func(), error) {
	m.mu.Lock()
	sem := m.sem
	m.mu.Unlock()

	select {
	case sem <- struct{}{}:
		return func() { <-sem }, nil
	case <-ctx.Done():
		return nil, ctx.Err()
	}
}

func (m *Manager) syncer() *syncer.Syncer {
	m.mu.Lock()
	defer m.mu.Unlock()
	return m.sync
}

func (m *Manager) update(name string, fn func(*Status)) {
	m.mu.Lock()
	defer m.mu.Unlock()
	if s, ok := m.stats[name]; ok {
		fn(s)
	}
}

// prepareDirs creates the mirror store and the isolated HOME git/ssh will use.
func prepareDirs(workDir string) (home string, err error) {
	home = filepath.Join(workDir, "home")
	for _, d := range []string{filepath.Join(workDir, "mirrors"), home} {
		if err := os.MkdirAll(d, 0o700); err != nil {
			return "", fmt.Errorf("prepare %s: %w", d, err)
		}
	}
	return home, nil
}

// backoff grows the retry delay geometrically, capped at max, so a repo that
// is down for hours stops hammering it (and our logs) every interval.
func backoff(interval, max time.Duration, failures int) time.Duration {
	d := interval
	for i := 1; i < failures && d < max; i++ {
		d *= 2
	}
	return min(d, max)
}

// stagger spreads the first run of each repo so a restart with many repos does
// not fire every git process in the same instant.
func stagger(index int) time.Duration {
	return min(time.Duration(index)*250*time.Millisecond, 15*time.Second)
}