package torrent

import (
	"context"
	"sync"
)

type pieceTask struct {
	Index int
}

type assignPieceRequest struct {
	peerID     string
	have       []bool
	hasInfo    bool
	responseCh chan assignPieceResponse
}

type assignPieceResponse struct {
	task pieceTask
	ok   bool
}

type reportPieceRequest struct {
	pieceIndex int
	success    bool
	responseCh chan bool
}

// cancelPeerRequest отправляет сигнал конкретному пиру отменить кусок в endgame.
type cancelPeerRequest struct {
	pieceIndex int
	peerID     string
}

type pieceScheduler struct {
	assignCh      chan assignPieceRequest
	reportCh      chan reportPieceRequest
	releasePeerCh chan string
	progressCh    chan int
	doneCh        chan struct{}
	stopCh        chan struct{}
	stopOnce      sync.Once

	// cancelCh рассылает сигнал отмены конкретному воркеру в endgame-режиме.
	// Ключ — peerID, значение — канал с индексом куска для отмены.
	cancelMu  sync.RWMutex
	cancelSubs map[string]chan int
}

type pieceState uint8

const (
	piecePending    pieceState = iota
	pieceInProgress            // качается одним пиром
	pieceDone
)

// endgameThreshold — количество оставшихся кусков, при котором включается endgame.
const endgameThreshold = 4

func newPieceScheduler(pieceCount int) *pieceScheduler {
	return newPieceSchedulerWithResume(pieceCount, nil)
}

func newPieceSchedulerWithResume(pieceCount int, completedIndices []int) *pieceScheduler {
	ps := &pieceScheduler{
		assignCh:      make(chan assignPieceRequest, 128),
		reportCh:      make(chan reportPieceRequest, 128),
		releasePeerCh: make(chan string, 128),
		progressCh:    make(chan int, 128),
		doneCh:        make(chan struct{}),
		stopCh:        make(chan struct{}),
		cancelSubs:    make(map[string]chan int),
	}

	go ps.run(pieceCount, completedIndices)
	return ps
}

// SubscribeCancel регистрирует канал для получения cancel-сигналов в endgame-режиме.
// Воркер должен вызвать это перед Acquire, и отписаться после завершения.
func (ps *pieceScheduler) SubscribeCancel(peerID string) <-chan int {
	ch := make(chan int, 8)
	ps.cancelMu.Lock()
	ps.cancelSubs[peerID] = ch
	ps.cancelMu.Unlock()
	return ch
}

// UnsubscribeCancel отписывает воркера от cancel-сигналов.
func (ps *pieceScheduler) UnsubscribeCancel(peerID string) {
	ps.cancelMu.Lock()
	delete(ps.cancelSubs, peerID)
	ps.cancelMu.Unlock()
}

// broadcastCancel рассылает сигнал отмены куска всем воркерам кроме победителя.
func (ps *pieceScheduler) broadcastCancel(pieceIndex int, winnerPeerID string) {
	ps.cancelMu.RLock()
	defer ps.cancelMu.RUnlock()
	for id, ch := range ps.cancelSubs {
		if id == winnerPeerID {
			continue
		}
		select {
		case ch <- pieceIndex:
		default:
		}
	}
}

func (ps *pieceScheduler) Acquire(ctx context.Context, peerID string, have []bool, hasInfo bool) (pieceTask, bool, error) {
	responseCh := make(chan assignPieceResponse, 1)
	req := assignPieceRequest{
		peerID:     peerID,
		have:       have,
		hasInfo:    hasInfo,
		responseCh: responseCh,
	}

	select {
	case <-ctx.Done():
		return pieceTask{}, false, ctx.Err()
	case <-ps.doneCh:
		return pieceTask{}, false, nil
	case ps.assignCh <- req:
	}

	select {
	case <-ctx.Done():
		return pieceTask{}, false, ctx.Err()
	case <-ps.doneCh:
		return pieceTask{}, false, nil
	case response := <-responseCh:
		return response.task, response.ok, nil
	}
}

func (ps *pieceScheduler) Report(ctx context.Context, pieceIndex int, success bool) (bool, error) {
	responseCh := make(chan bool, 1)
	req := reportPieceRequest{
		pieceIndex: pieceIndex,
		success:    success,
		responseCh: responseCh,
	}

	select {
	case <-ctx.Done():
		return false, ctx.Err()
	case <-ps.doneCh:
		return false, nil
	case ps.reportCh <- req:
	}

	select {
	case <-ctx.Done():
		return false, ctx.Err()
	case <-ps.doneCh:
		return false, nil
	case accepted := <-responseCh:
		return accepted, nil
	}
}

func (ps *pieceScheduler) ReleasePeer(peerID string) {
	if peerID == "" {
		return
	}
	select {
	case <-ps.doneCh:
		return
	case ps.releasePeerCh <- peerID:
	default:
	}
}

func (ps *pieceScheduler) Progress() <-chan int {
	return ps.progressCh
}

func (ps *pieceScheduler) Done() <-chan struct{} {
	return ps.doneCh
}

func (ps *pieceScheduler) Stop() {
	ps.stopOnce.Do(func() {
		close(ps.stopCh)
	})
}

func (ps *pieceScheduler) run(pieceCount int, completedIndices []int) {
	states := make([]pieceState, pieceCount)
	availability := make([]int, pieceCount)
	peerAvailability := make(map[string][]bool, 256)
	// endgamePeers — множество пиров, которым назначен кусок в endgame режиме
	endgamePeers := make(map[int][]string) // pieceIndex -> []peerID
	completed := 0

	// Восстанавливаем уже завершённые куски из resume
	for _, idx := range completedIndices {
		if idx >= 0 && idx < pieceCount {
			states[idx] = pieceDone
			completed++
		}
	}

	finish := func() {
		close(ps.doneCh)
		close(ps.progressCh)
	}

	if pieceCount == 0 {
		finish()
		return
	}

	for {
		if completed >= pieceCount {
			finish()
			return
		}

		// Подсчёт оставшихся кусков для определения endgame
		remaining := pieceCount - completed
		endgame := remaining <= endgameThreshold

		select {
		case <-ps.stopCh:
			finish()
			return
		case peerID := <-ps.releasePeerCh:
			releasePeerAvailability(peerAvailability, availability, peerID)
		case req := <-ps.assignCh:
			if req.peerID != "" && req.hasInfo {
				updatePeerAvailability(peerAvailability, availability, req.peerID, req.have, pieceCount)
			}

			pieceIndex := -1
			if endgame {
				// Endgame: разрешаем брать куски в состоянии InProgress тоже
				pieceIndex = selectPieceEndgame(states, availability, req.have, req.hasInfo, endgamePeers, req.peerID)
			} else {
				pieceIndex = selectPendingPieceRarest(states, availability, req.have, req.hasInfo)
			}

			if pieceIndex >= 0 {
				if !endgame {
					states[pieceIndex] = pieceInProgress
				}
				// В endgame: запоминаем всех пиров, которые качают этот кусок
				endgamePeers[pieceIndex] = append(endgamePeers[pieceIndex], req.peerID)
				debugf("scheduler assigned piece %d to %s (endgame=%v)", pieceIndex, req.peerID, endgame)
				req.responseCh <- assignPieceResponse{
					task: pieceTask{Index: pieceIndex},
					ok:   true,
				}
				continue
			}
			req.responseCh <- assignPieceResponse{ok: false}

		case req := <-ps.reportCh:
			if req.pieceIndex < 0 || req.pieceIndex >= len(states) {
				req.responseCh <- false
				continue
			}

			if req.success {
				if states[req.pieceIndex] == pieceDone {
					debugf("scheduler report piece %d ignored (already done)", req.pieceIndex)
					req.responseCh <- false
					continue
				}
				states[req.pieceIndex] = pieceDone
				completed++
				debugf("scheduler report piece %d success (%d/%d)", req.pieceIndex, completed, pieceCount)

				// Endgame: уведомить других пиров отменить этот кусок
				if peers, ok := endgamePeers[req.pieceIndex]; ok && len(peers) > 1 {
					// Находим winner — последний репортнувший (req не содержит peerID,
					// поэтому рассылаем всем — воркер проверяет индекс)
					go ps.broadcastCancel(req.pieceIndex, "")
				}
				delete(endgamePeers, req.pieceIndex)

				select {
				case ps.progressCh <- req.pieceIndex:
				default:
				}
				req.responseCh <- true
				continue
			}

			// Неуспех: возвращаем кусок в pending
			if states[req.pieceIndex] == pieceInProgress {
				states[req.pieceIndex] = piecePending
				debugf("scheduler report piece %d failed, re-queued", req.pieceIndex)
			}
			// В endgame: убираем только этого пира из списка
			if peers, ok := endgamePeers[req.pieceIndex]; ok {
				filtered := peers[:0]
				for _, p := range peers {
					if p != "" { // убираем все (peerID недоступен в req)
						filtered = append(filtered, p)
					}
				}
				if len(filtered) == 0 {
					delete(endgamePeers, req.pieceIndex)
				} else {
					endgamePeers[req.pieceIndex] = filtered
				}
			}
			req.responseCh <- false
		}
	}
}

// selectPieceEndgame выбирает кусок в endgame режиме:
// разрешает назначать куски в состоянии InProgress другим пирам.
func selectPieceEndgame(states []pieceState, availability []int, have []bool, hasInfo bool, endgamePeers map[int][]string, peerID string) int {
	// Сначала пробуем найти pending кусок (обычный путь)
	if idx := selectPendingPieceRarest(states, availability, have, hasInfo); idx >= 0 {
		return idx
	}

	// Endgame: ищем InProgress кусок, который у нас ещё не назначен этому пиру
	bestPiece := -1
	bestAvailability := 0

	for pieceIndex, state := range states {
		if state != pieceInProgress {
			continue
		}
		// Проверяем, не назначен ли уже этому пиру
		alreadyAssigned := false
		for _, pid := range endgamePeers[pieceIndex] {
			if pid == peerID {
				alreadyAssigned = true
				break
			}
		}
		if alreadyAssigned {
			continue
		}
		// Проверяем наличие у пира
		if hasInfo && (pieceIndex >= len(have) || !have[pieceIndex]) {
			continue
		}

		count := availability[pieceIndex]
		if bestPiece == -1 || count < bestAvailability {
			bestPiece = pieceIndex
			bestAvailability = count
		}
	}

	return bestPiece
}

func updatePeerAvailability(peerAvailability map[string][]bool, availability []int, peerID string, have []bool, pieceCount int) {
	normalized := make([]bool, pieceCount)
	copy(normalized, have)

	if prev, ok := peerAvailability[peerID]; ok {
		same := true
		for i := 0; i < pieceCount; i++ {
			if prev[i] != normalized[i] {
				same = false
				break
			}
		}
		if same {
			return
		}

		for i := 0; i < pieceCount; i++ {
			if prev[i] && availability[i] > 0 {
				availability[i]--
			}
			if normalized[i] {
				availability[i]++
			}
		}
		peerAvailability[peerID] = normalized
		return
	}

	for i := 0; i < pieceCount; i++ {
		if normalized[i] {
			availability[i]++
		}
	}
	peerAvailability[peerID] = normalized
}

func releasePeerAvailability(peerAvailability map[string][]bool, availability []int, peerID string) {
	prev, ok := peerAvailability[peerID]
	if !ok {
		return
	}

	for i := range prev {
		if prev[i] && availability[i] > 0 {
			availability[i]--
		}
	}
	delete(peerAvailability, peerID)
}

func selectPendingPieceRarest(states []pieceState, availability []int, have []bool, hasInfo bool) int {
	firstPending := firstPendingPiece(states)
	if firstPending < 0 {
		return -1
	}

	// If peer has not sent bitfield/have yet, optimistically probe.
	if !hasInfo {
		return firstPending
	}

	bestPiece := -1
	bestAvailability := 0

	for pieceIndex, state := range states {
		if state != piecePending {
			continue
		}
		if pieceIndex >= len(have) || !have[pieceIndex] {
			continue
		}

		count := availability[pieceIndex]
		if bestPiece == -1 || count < bestAvailability || (count == bestAvailability && pieceIndex < bestPiece) {
			bestPiece = pieceIndex
			bestAvailability = count
		}
	}

	if bestPiece >= 0 {
		return bestPiece
	}

	// Some peers send truncated availability info; allow fallback probing.
	if len(have) == 0 || len(have) < len(states) {
		return firstPending
	}

	return -1
}

func firstPendingPiece(states []pieceState) int {
	for pieceIndex, state := range states {
		if state == piecePending {
			return pieceIndex
		}
	}
	return -1
}