package appview

import (
	"context"
	"errors"
	"path/filepath"
	"testing"
	"time"

	"provenance/internal/domain"
	"provenance/internal/storage"
	"provenance/internal/trust"
)

type reconnectStream struct {
	cursors chan int64
	change  domain.Change
}

func (f *reconnectStream) Consume(ctx context.Context, cursor int64, wanted func(context.Context, string, int64) (bool, error), apply func(context.Context, domain.Change) error) error {
	f.cursors <- cursor
	if f.change.TimeUS == 0 {
		f.change.TimeUS = cursor + time.Second.Microseconds()
		selected, err := wanted(ctx, f.change.DID, f.change.TimeUS)
		if err != nil || !selected {
			return errors.New("participant filter rejected the event")
		}
		if err := apply(ctx, f.change); err != nil {
			return err
		}
		return errors.New("connection closed")
	}
	<-ctx.Done()
	return ctx.Err()
}

func TestReconnectUsesCommittedCursor(t *testing.T) {
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	store, err := storage.Open(filepath.Join(t.TempDir(), "index.db"))
	if err != nil {
		t.Fatal(err)
	}
	defer store.Close()
	repo := &repository{}
	service := New(store, repo, trust.Limits{States: 100, Edges: 100, Penalty: .1})
	if err := service.Join(ctx, alice); err != nil {
		t.Fatal(err)
	}
	v := domain.Vouch{Author: alice, Subject: bob, Score: 2}
	stream := &reconnectStream{cursors: make(chan int64, 2), change: domain.Change{Kind: domain.RecordChange, DID: alice, RecordKey: domain.RecordKey(bob, ""), Vouch: &v}}
	done := make(chan struct{})
	go func() { defer close(done); service.Run(ctx, stream) }()
	var first, second int64
	select {
	case first = <-stream.cursors:
	case <-ctx.Done():
		t.Fatal("stream did not start")
	}
	select {
	case second = <-stream.cursors:
	case <-ctx.Done():
		t.Fatal("stream did not reconnect")
	}
	cancel()
	<-done
	if second != first+time.Second.Microseconds() {
		t.Fatalf("reconnect cursor = %d, initial = %d", second, first)
	}
	result, err := service.Reputation(context.Background(), alice, bob, "", domain.Default)
	if err != nil || result.Score == nil || *result.Score != 2 {
		t.Fatal("stream event did not reach the graph")
	}
}

type blockingRepository struct {
	repository
	entered chan struct{}
	release chan struct{}
	vouch   domain.Vouch
}

func (r *blockingRepository) Snapshot(ctx context.Context, _ string) ([]domain.Vouch, error) {
	close(r.entered)
	select {
	case <-r.release:
		return []domain.Vouch{r.vouch}, nil
	case <-ctx.Done():
		return nil, ctx.Err()
	}
}

func TestBootstrapAndStreamUpdate(t *testing.T) {
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	store, err := storage.Open(filepath.Join(t.TempDir(), "index.db"))
	if err != nil {
		t.Fatal(err)
	}
	defer store.Close()
	v := domain.Vouch{Author: alice, Subject: bob, Score: 2}
	repo := &blockingRepository{entered: make(chan struct{}), release: make(chan struct{}), vouch: v}
	service := New(store, repo, trust.Limits{States: 100, Edges: 100, Penalty: .1})
	joined := make(chan error, 1)
	go func() { joined <- service.Join(ctx, alice) }()
	<-repo.entered
	v.Score = domain.Blocked
	applied := make(chan error, 1)
	go func() {
		applied <- service.Apply(ctx, domain.Change{Kind: domain.RecordChange, DID: alice, TimeUS: time.Now().Add(time.Second).UnixMicro(), RecordKey: domain.RecordKey(bob, ""), Vouch: &v})
	}()
	close(repo.release)
	if err := <-joined; err != nil {
		t.Fatal(err)
	}
	if err := <-applied; err != nil {
		t.Fatal(err)
	}
	result, err := service.Reputation(ctx, alice, bob, "", domain.Default)
	if err != nil || result.Score == nil || *result.Score != domain.Blocked {
		t.Fatal("bootstrap overwrote a concurrent block")
	}
}