package microcosm

import (
	"context"
	"encoding/json"
	"errors"
	"net/http"
	"net/http/httptest"
	"strconv"
	"strings"
	"testing"

	"github.com/coder/websocket"

	"provenance/internal/atproto"
	"provenance/internal/domain"
)

const author = "did:plc:alice"
const subject = "did:plc:bob"
const recordCID = "bafyreidtest"

func commit(t *testing.T, operation Operation, record any) []byte {
	t.Helper()
	value := map[string]any{"did": author, "time_us": int64(10000000), "kind": "commit", "commit": map[string]any{"operation": operation, "collection": domain.Collection, "rkey": domain.RecordKey(subject, "work"), "record": record, "cid": recordCID}}
	data, err := json.Marshal(value)
	if err != nil {
		t.Fatal(err)
	}
	return data
}

func validRecord() atproto.Record {
	score := 2
	return atproto.Record{Type: domain.Collection, Subject: subject, Tag: "work", Score: &score, CreatedAt: "2026-10-07T00:00:00Z"}
}

func TestWebSocketStream(t *testing.T) {
	stop := errors.New("complete")
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if r.URL.Query().Get("wantedCollections") != domain.Collection || r.URL.Query().Get("cursor") != strconv.FormatInt(10000000-ReplayOverlap.Microseconds(), 10) {
			t.Error("wrong collection filter or replay cursor")
		}
		conn, err := websocket.Accept(w, r, nil)
		if err != nil {
			t.Error(err)
			return
		}
		defer conn.CloseNow()
		for _, operation := range []Operation{Create, Update, Delete} {
			if err := conn.Write(r.Context(), websocket.MessageText, commit(t, operation, validRecord())); err != nil {
				return
			}
		}
		conn.Read(r.Context())
	}))
	defer server.Close()
	client, err := New("ws"+strings.TrimPrefix(server.URL, "http")+"/subscribe", server.URL)
	if err != nil {
		t.Fatal(err)
	}
	var changes []domain.Change
	err = client.Consume(context.Background(), 10000000, nil, func(_ context.Context, change domain.Change) error {
		changes = append(changes, change)
		if len(changes) == 3 {
			return stop
		}
		return nil
	})
	if !errors.Is(err, stop) || len(changes) != 3 {
		t.Fatalf("received %d changes: %v", len(changes), err)
	}
	if changes[0].Vouch == nil || changes[0].Vouch.Score != 2 || changes[2].Vouch != nil {
		t.Fatalf("incorrect commit mapping: %+v", changes)
	}
}

func TestSlingshotHydration(t *testing.T) {
	calls := 0
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		calls++
		if r.URL.Path != "/xrpc/com.atproto.repo.getRecord" || r.URL.Query().Get("repo") != author || r.URL.Query().Get("cid") != recordCID {
			t.Error("wrong Slingshot query")
		}
		json.NewEncoder(w).Encode(map[string]any{"uri": "at://" + author + "/" + domain.Collection + "/" + domain.RecordKey(subject, "work"), "cid": recordCID, "value": validRecord()})
	}))
	defer server.Close()
	client, err := New("ws://127.0.0.1:1/subscribe", server.URL)
	if err != nil {
		t.Fatal(err)
	}
	change, err := client.decode(context.Background(), commit(t, Create, nil), nil)
	if err != nil || change.Vouch == nil || calls != 1 {
		t.Fatalf("hydration: %+v %v", change, err)
	}
	change, err = client.decode(context.Background(), commit(t, Create, nil), func(context.Context, string, int64) (bool, error) { return false, nil })
	if err != nil || change.Vouch != nil || calls != 1 {
		t.Fatal("non-participant triggered record hydration")
	}
}

func TestInvalidUpdateRemovesAssertion(t *testing.T) {
	client, err := New(DefaultJetstreamURL, DefaultSlingshotURL)
	if err != nil {
		t.Fatal(err)
	}
	record := validRecord()
	score := domain.SelfScore
	record.Score = &score
	change, err := client.decode(context.Background(), commit(t, Update, record), nil)
	if err != nil || change.Kind != domain.RecordChange || change.Vouch != nil || change.RecordKey == "" {
		t.Fatalf("invalid update retained its assertion: %+v %v", change, err)
	}
	if _, err := client.decode(context.Background(), []byte(`{"kind":"commit"}`), nil); err == nil {
		t.Fatal("accepted malformed event envelope")
	}
}

func TestNonCanonicalRecords(t *testing.T) {
	wrongSubject := validRecord()
	wrongSubject.Subject = "did:plc:carol"
	wrongTag := validRecord()
	wrongTag.Tag = "Work"
	for _, tt := range []struct {
		name   string
		record atproto.Record
	}{
		{"subject mismatch", wrongSubject}, {"tag mismatch", wrongTag},
	} {
		t.Run(tt.name, func(t *testing.T) {
			server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
				key := domain.RecordKey(subject, "work")
				if r.URL.Query().Get("rkey") != key {
					t.Error("wrong hydration record key")
				}
				json.NewEncoder(w).Encode(map[string]any{"uri": "at://" + author + "/" + domain.Collection + "/" + key, "cid": recordCID, "value": tt.record})
			}))
			defer server.Close()
			client, err := New("ws://127.0.0.1:1/subscribe", server.URL)
			if err != nil {
				t.Fatal(err)
			}
			for _, operation := range []Operation{Create, Update} {
				for _, payload := range []any{tt.record, nil} {
					change, err := client.decode(context.Background(), commit(t, operation, payload), nil)
					if err != nil || change.Vouch != nil || change.RecordKey != domain.RecordKey(subject, "work") || change.Kind != domain.RecordChange {
						t.Fatalf("non-canonical %s entered the index: %+v %v", operation, change, err)
					}
				}
			}
		})
	}
}

func TestEndpointValidation(t *testing.T) {
	for _, address := range []string{"ws://remote.example.com/subscribe", "wss://user:pass@example.com/subscribe", "wss://example.com/subscribe?compress=true"} {
		if _, err := New(address, DefaultSlingshotURL); err == nil {
			t.Fatalf("accepted %s", address)
		}
	}
	if _, err := New("wss://stream.internal.example/subscribe", "https://records.internal.example"); err != nil {
		t.Fatal("rejected configured self-hosted endpoints")
	}
}