package replica

import (
	"context"
	"errors"
	"net"
	"strconv"
	"sync"
	"sync/atomic"
	"time"

	"github.com/go-git/go-git/v5"
	"github.com/go-git/go-git/v5/plumbing"
	"github.com/go-git/go-git/v5/plumbing/transport"
)

var (
	ErrFetchIdle  = errors.New("replica: whole-history fetch idle bound exceeded")
	ErrFetchTotal = errors.New("replica: whole-history fetch total bound exceeded")
	activeFetches sync.Map
	fetchSequence atomic.Uint64
)

type fetchLimit struct {
	ctx   context.Context
	idle  time.Duration
	total time.Duration
	timer *time.Timer
}

type fetchConn struct {
	net.Conn
	limit *fetchLimit
	stop  func() bool
}

func (c *fetchConn) Read(p []byte) (int, error) {
	n, err := c.Conn.Read(p)
	if n > 0 && c.limit.timer != nil {
		c.limit.timer.Reset(c.limit.idle)
	}
	return n, err
}

func (c *fetchConn) Write(p []byte) (int, error) {
	n, err := c.Conn.Write(p)
	if n > 0 && c.limit.timer != nil {
		c.limit.timer.Reset(c.limit.idle)
	}
	return n, err
}

func (c *fetchConn) Close() error {
	c.stop()
	return c.Conn.Close()
}

func FetchHistory(ctx context.Context, repo *git.Repository, url string, auth transport.AuthMethod, branch plumbing.ReferenceName) (plumbing.Hash, error) {
	a, ok := auth.(*SSH)
	if !ok {
		return Fetch(ctx, repo, url, auth, branch)
	}
	idle, total := a.FetchIdleTimeout, a.FetchTotalTimeout
	if idle <= 0 {
		idle = DefaultFetchIdleTimeout
	}
	if total <= 0 {
		total = DefaultFetchTotalTimeout
	}
	ctx, cancel := context.WithCancelCause(ctx)
	defer cancel(nil)
	limit := &fetchLimit{ctx: ctx, idle: idle}
	limit.timer = time.AfterFunc(idle, func() { cancel(ErrFetchIdle) })
	defer limit.timer.Stop()
	totalTimer := time.AfterFunc(total, func() { cancel(ErrFetchTotal) })
	defer totalTimer.Stop()
	copy := *a
	copy.Timeout = total
	auth, release := registerFetch(&copy, limit)
	defer release()
	h, err := Fetch(ctx, repo, url, auth, branch)
	if cause := context.Cause(ctx); cause != nil {
		return plumbing.ZeroHash, cause
	}
	return h, err
}

func registerFetch(auth *SSH, limit *fetchLimit) (*SSH, func()) {
	copy := *auth
	copy.fetch = strconv.FormatUint(fetchSequence.Add(1), 10)
	activeFetches.Store(copy.fetch, limit)
	return &copy, func() { activeFetches.Delete(copy.fetch) }
}