package pipeline

import (
	"context"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"evobgp/internal/store"
)

type blockingPrune struct {
	store.Backend
	calls   atomic.Int32
	started chan struct{}
	release chan struct{}
	once    sync.Once
}

func (b *blockingPrune) PruneRevisionsBefore(ctx context.Context, tenantID string, cutoff time.Time) (int, error) {
	n := b.calls.Add(1)
	if n == 1 {
		b.once.Do(func() { close(b.started) })
		select {
		case <-b.release:
		case <-ctx.Done():
			return 0, ctx.Err()
		}
	}
	return b.Backend.PruneRevisionsBefore(ctx, tenantID, cutoff)
}

func TestRenderDoesNotWaitForRevisionPrune(t *testing.T) {
	m := store.NewMemory()
	m.SeedDemo()
	tenant, _, modIP, _, _ := m.DemoIDs()
	release := make(chan struct{})
	var releaseOnce sync.Once
	closeRelease := func() { releaseOnce.Do(func() { close(release) }) }
	t.Cleanup(closeRelease)
	st := &blockingPrune{Backend: m, started: make(chan struct{}), release: release}
	var shift atomic.Int64
	revisionPrunes.mu.Lock()
	prevNow := revisionPrunes.now
	revisionPrunes.now = func() time.Time {
		return time.Now().Add(time.Duration(shift.Load()) * time.Minute)
	}
	revisionPrunes.mu.Unlock()
	t.Cleanup(func() {
		revisionPrunes.mu.Lock()
		revisionPrunes.now = prevNow
		revisionPrunes.mu.Unlock()
	})

	rows := []store.PrefixRow{{Prefix: "203.0.113.55/32", Source: "ip_range"}}
	renderDone := make(chan error, 1)
	go func() {
		_, err := RenderTenantRevisionFromPrefixes(context.Background(), st, nil, tenant, modIP, rows)
		renderDone <- err
	}()
	select {
	case err := <-renderDone:
		if err != nil {
			t.Fatal(err)
		}
	case <-time.After(5 * time.Second):
		t.Fatal("render waited for revision prune")
	}
	select {
	case <-st.started:
	case <-time.After(5 * time.Second):
		t.Fatal("prune did not start")
	}
	if st.calls.Load() != 1 {
		t.Fatalf("prune calls=%d", st.calls.Load())
	}
	applyRevisionRetention(st, tenant)
	if st.calls.Load() != 1 {
		t.Fatalf("overlapping render stacked prune calls=%d", st.calls.Load())
	}
	closeRelease()
	waitCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()
	if err := waitRevisionPrune(waitCtx, tenant); err != nil {
		t.Fatal(err)
	}
	shift.Store(10)
	applyRevisionRetention(st, tenant)
	waitCtx2, cancel2 := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel2()
	if err := waitRevisionPrune(waitCtx2, tenant); err != nil {
		t.Fatal(err)
	}
	if st.calls.Load() != 2 {
		t.Fatalf("prune after interval calls=%d", st.calls.Load())
	}
}

func TestPruneTenantRevisionsDirect(t *testing.T) {
	m := store.NewMemory()
	m.SeedDemo()
	tenant, _, _, _, _ := m.DemoIDs()
	var calls atomic.Int32
	st := pruneCountBackend{Backend: m, calls: &calls}
	pruneTenantRevisions(context.Background(), st, tenant)
	if calls.Load() != 1 {
		t.Fatalf("calls=%d", calls.Load())
	}
}

type pruneCountBackend struct {
	store.Backend
	calls *atomic.Int32
}

func (p pruneCountBackend) PruneRevisionsBefore(ctx context.Context, tenantID string, cutoff time.Time) (int, error) {
	p.calls.Add(1)
	return p.Backend.PruneRevisionsBefore(ctx, tenantID, cutoff)
}