//go:build interop

package interop

import (
	"bytes"
	"context"
	"encoding/base64"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"os"
	"os/exec"
	"path/filepath"
	"strings"
	"testing"
	"time"

	"github.com/bluesky-social/indigo/atproto/identity"
	"github.com/bluesky-social/indigo/atproto/syntax"
	indigoxrpc "github.com/bluesky-social/indigo/xrpc"
	"github.com/gorilla/websocket"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	"tangled.org/core/knotfeed"
	"tangled.org/core/tapc"
)

const seedRoute = "/xrpc/org.tangled.temp.server.seedRepo"
const fosterOwner = "did:plc:boltfulboy"
const fosterRkey = "the-feral-pearl-ledger"
const fosterCreatedAt = "1970-01-01T00:18:20Z"

func seedPost(t *testing.T, user, password string, body any) (int, string) {
	t.Helper()
	encoded, err := json.Marshal(body)
	require.NoError(t, err)
	request, err := http.NewRequest(
		http.MethodPost,
		os.Getenv("KNOT_TEST_BASE_URL")+seedRoute,
		bytes.NewReader(encoded),
	)
	require.NoError(t, err)
	request.Header.Set("Content-Type", "application/json")
	if user != "" {
		request.SetBasicAuth(user, password)
	}
	client := &http.Client{Timeout: 30 * time.Second}
	response, err := client.Do(request)
	require.NoError(t, err)
	defer response.Body.Close()
	raw, err := io.ReadAll(response.Body)
	require.NoError(t, err)
	return response.StatusCode, string(raw)
}

func seedRequest(t *testing.T) (map[string]any, syntax.RecordKey) {
	t.Helper()
	rkey, err := syntax.ParseRecordKey(fosterRkey)
	require.NoError(t, err, "fosterRkey must be a record key")
	fosterRepoDid := os.Getenv("KNOT_TEST_FOSTER_REPO_DID")
	require.NotEmpty(
		t,
		fosterRepoDid,
		"KNOT_TEST_FOSTER_REPO_DID is the foreign repo DID the knot fosters",
	)
	return map[string]any{
		"repoDid": fosterRepoDid,
		"manifest": map[string]any{
			"$type":       "org.tangled.repo.manifest",
			"declaration": map[string]any{"owner": fosterOwner, "name": "boltfulboy"},
		},
		"records": []map[string]any{{
			"collection": ticketCollection,
			"rkey":       rkey.String(),
			"author":     fosterOwner,
			"createdAt":  1100,
			"record": map[string]any{
				"$type":         ticketCollection,
				"title":         map[string]any{"$type": "org.tangled.markup.markdown#inline", "text": "a feral pearl"},
				"state":         "open",
				"createdAt":     fosterCreatedAt,
				"x-tngl-editor": fosterOwner,
			},
		}},
	}, rkey
}

func pushRepo(t *testing.T, source, addr, did, token string) {
	t.Helper()
	header := base64.StdEncoding.EncodeToString([]byte("x-tangled-token:" + token))
	command := exec.Command(
		"git", "-C", source,
		"-c", "http.extraHeader=Authorization: Basic "+header,
		"push", "-q", "--tags",
		fmt.Sprintf("http://%s/%s", addr, did),
		"main",
	)
	command.Env = append(os.Environ(), "GIT_TERMINAL_PROMPT=0")
	output, err := command.CombinedOutput()
	require.NoError(t, err, "the receive-pack push must deliver the staged history: %s", output)
}

func cloneRepo(t *testing.T, url string) string {
	t.Helper()
	clone := filepath.Join(t.TempDir(), "clone")
	cloneCommand := exec.Command("git", "clone", "-q", url, clone)
	cloneCommand.Env = append(os.Environ(), "GIT_TERMINAL_PROMPT=0")
	output, err := cloneCommand.CombinedOutput()
	require.NoError(t, err, "git clone must fetch the fostered repo: %s", output)
	return clone
}

func gitIn(t *testing.T, dir string, args ...string) string {
	t.Helper()
	output, err := exec.Command("git", append([]string{"-C", dir}, args...)...).CombinedOutput()
	require.NoError(t, err, "git %v failed: %s", args, output)
	return strings.TrimSpace(string(output))
}

func TestASeededKnotFostersAForeignRepoVerbatim(t *testing.T) {
	base := os.Getenv("KNOT_TEST_BASE_URL")
	require.NotEmpty(t, base, "KNOT_TEST_BASE_URL is the knot under test")
	addr := os.Getenv("KNOT_TEST_ADDR")
	require.NotEmpty(t, addr, "KNOT_TEST_ADDR is the knot's host:port")
	secret := os.Getenv("KNOT_TEST_FOSTER_SECRET")
	require.NotEmpty(t, secret, "KNOT_TEST_FOSTER_SECRET is the foster secret")
	source := os.Getenv("KNOT_TEST_FOSTER_SOURCE")
	require.NotEmpty(t, source, "KNOT_TEST_FOSTER_SOURCE is the staged source repo")
	wantSha := os.Getenv("KNOT_TEST_SHA")
	require.NotEmpty(t, wantSha, "KNOT_TEST_SHA is the staged source tip")
	pushToken := os.Getenv("KNOT_TEST_PUSH_TOKEN")
	require.NotEmpty(t, pushToken, "KNOT_TEST_PUSH_TOKEN authorizes the receive-pack push")

	ctx := context.Background()
	xc := knotClient(t)
	body, rkey := seedRequest(t)

	wrongPasswordStatus, wrongPasswordBody := seedPost(t, "admin", "not-the-secret", body)
	wrongUserStatus, wrongUserBody := seedPost(t, "root", secret, body)
	noHeaderStatus, noHeaderBody := seedPost(t, "", "", body)
	for _, attempt := range []struct {
		status int
		name   string
	}{
		{wrongPasswordStatus, "a wrong password"},
		{wrongUserStatus, "a wrong user"},
		{noHeaderStatus, "no header"},
	} {
		require.Equal(t, http.StatusUnauthorized, attempt.status, "%s is rejected", attempt.name)
	}
	assert.Equal(t, wrongPasswordBody, wrongUserBody, "a wrong password and a wrong user are indistinguishable")
	assert.Equal(t, wrongPasswordBody, noHeaderBody, "a missing header is indistinguishable too")

	status, raw := seedPost(t, "admin", secret, body)
	require.Equal(t, http.StatusOK, status, "the seed request must succeed: %s", raw)
	var seeded struct {
		Did     string `json:"did"`
		Key     string `json:"key"`
		Results []struct {
			Collection string `json:"collection"`
			Rkey       string `json:"rkey"`
			Error      string `json:"error"`
		} `json:"results"`
	}
	require.NoError(t, json.Unmarshal([]byte(raw), &seeded))
	require.Equal(
		t,
		os.Getenv("KNOT_TEST_FOSTER_REPO_DID"),
		seeded.Did,
		"the response echoes the supplied repo DID, never a minted one",
	)
	require.NotEmpty(t, seeded.Key, "the response reports the repo's signing key")
	require.Len(t, seeded.Results, 1, "one record was seeded")
	assert.Equal(t, ticketCollection, seeded.Results[0].Collection)
	assert.Equal(t, rkey.String(), seeded.Results[0].Rkey)
	assert.Empty(t, seeded.Results[0].Error, "the verbatim record seeds without lexicon validation")

	did, err := syntax.ParseDID(seeded.Did)
	require.NoError(t, err)
	collection, err := syntax.ParseNSID(ticketCollection)
	require.NoError(t, err)

	var listed struct {
		Records []struct {
			Uri   string `json:"uri"`
			Value struct {
				State     string `json:"state"`
				CreatedAt string `json:"createdAt"`
				Editor    string `json:"x-tngl-editor"`
			} `json:"value"`
		} `json:"records"`
	}
	err = xc.LexDo(ctx, indigoxrpc.Query, "", "com.atproto.repo.listRecords", map[string]any{
		"repo":       did.String(),
		"collection": ticketCollection,
		"limit":      50,
	}, nil, &listed)
	require.NoError(t, err, "repo.listRecords serves the fostered repo")
	require.Len(t, listed.Records, 1, "exactly one ticket is listed")
	assert.Equal(t, "at://"+did.String()+"/"+ticketCollection+"/"+rkey.String(), listed.Records[0].Uri)
	assert.Equal(t, "feral", listed.Records[0].Value.State, "the lexicon-invalid state is served verbatim")
	assert.Equal(t, fosterCreatedAt, listed.Records[0].Value.CreatedAt, "the 1970 createdAt is served verbatim")
	assert.Equal(t, fosterOwner, listed.Records[0].Value.Editor, "the foreign editor is served verbatim")

	var fetched struct {
		Uri   string `json:"uri"`
		Value struct {
			Title struct {
				Type string `json:"$type"`
				Text string `json:"text"`
			} `json:"title"`
		} `json:"value"`
	}
	err = xc.LexDo(ctx, indigoxrpc.Query, "", "com.atproto.repo.getRecord", map[string]any{
		"repo":       did.String(),
		"collection": ticketCollection,
		"rkey":       rkey.String(),
	}, nil, &fetched)
	require.NoError(t, err, "repo.getRecord serves the seeded record")
	assert.Equal(t, listed.Records[0].Uri, fetched.Uri)
	assert.Equal(t, "org.tangled.markup.markdown#inline", fetched.Value.Title.Type, "the title union is served in the org.tangled shape")
	assert.Equal(t, "a feral pearl", fetched.Value.Title.Text)

	document := fmt.Sprintf(
		`{"id":%q,"verificationMethod":[{"id":%q,"type":"Multikey","controller":%q,"publicKeyMultibase":%q}]}`,
		did.String(), did.String()+"#atproto", did.String(), seeded.Key,
	)
	var doc identity.DIDDocument
	require.NoError(t, json.Unmarshal([]byte(document), &doc), "the seed response includes the key material of a did document")
	dir := identity.NewMockDirectory()
	dir.Insert(identity.ParseIdentity(&doc))
	_, full, _ := provenRecord(t, ctx, xc, dir, did, collection, rkey)

	leaf, _, err := full.GetRecordBytes(ctx, collection, rkey)
	require.NoError(t, err, "the stored record bytes come back out of the signed tree")
	stored, unions := leafFields(t, leaf)
	assert.Equal(t, ticketCollection, stored["$type"])
	assert.Equal(t, "feral", stored["state"], "the cbor leaf keeps the state the validated write path would refuse")
	assert.Equal(t, fosterCreatedAt, stored["createdAt"], "the cbor leaf keeps the 1970 createdAt")
	assert.Equal(t, fosterOwner, stored["x-tngl-editor"], "the cbor leaf keeps the foreign editor")
	title, ok := unions["title"]
	require.True(t, ok, "the leaf stores the title as the inline markup union")
	assert.Equal(t, "org.tangled.markup.markdown#inline", title["$type"])
	assert.Equal(t, "a feral pearl", title["text"])

	pushRepo(t, source, addr, did.String(), pushToken)
	clone := cloneRepo(t, "http://"+addr+"/"+did.String())
	assert.Equal(t, wantSha, gitIn(t, clone, "rev-parse", "HEAD"), "the clone's HEAD is the pushed tip")
	assert.Equal(t, gitIn(t, source, "symbolic-ref", "--short", "HEAD"), gitIn(t, clone, "symbolic-ref", "--short", "HEAD"), "the clone keeps the source's default branch")
	assert.Equal(t, "v1.0.0", gitIn(t, clone, "tag", "--points-at", "HEAD"), "the annotated tag survives the round trip")
}

func TestFosteredReposAnnounceOnTheFirehose(t *testing.T) {
	base := os.Getenv("KNOT_TEST_BASE_URL")
	require.NotEmpty(t, base, "KNOT_TEST_BASE_URL is the knot under test")
	addr := os.Getenv("KNOT_TEST_ADDR")
	require.NotEmpty(t, addr, "KNOT_TEST_ADDR is the knot's host:port")
	secret := os.Getenv("KNOT_TEST_FOSTER_SECRET")
	require.NotEmpty(t, secret, "KNOT_TEST_FOSTER_SECRET is the foster secret")
	source := os.Getenv("KNOT_TEST_FOSTER_SOURCE")
	require.NotEmpty(t, source, "KNOT_TEST_FOSTER_SOURCE is the staged source repo")
	wantSha := os.Getenv("KNOT_TEST_SHA")
	require.NotEmpty(t, wantSha, "KNOT_TEST_SHA is the staged source tip")

	body, rkey := seedRequest(t)
	endpoint := "ws://" + strings.TrimPrefix(base, "http://") + "/xrpc/com.atproto.sync.subscribeRepos"
	conn, _, err := websocket.DefaultDialer.DialContext(context.Background(), endpoint, nil)
	require.NoError(t, err)
	defer conn.Close()

	status, raw := seedPost(t, "admin", secret, body)
	require.Equal(t, http.StatusOK, status, "the seed request must succeed: %s", raw)
	var seeded struct {
		Did string `json:"did"`
	}
	require.NoError(t, json.Unmarshal([]byte(raw), &seeded))
	did, err := syntax.ParseDID(seeded.Did)
	require.NoError(t, err)

	wantPath := ticketCollection + "/" + rkey.String()
	announced := false
	scan := time.Now().Add(15 * time.Second)
	for !announced && time.Now().Before(scan) {
		frame := nextFrame(t, conn)
		if frame.Commit == nil || frame.Commit.Repo != did.String() {
			continue
		}
		for _, op := range frame.Commit.Ops {
			if op.Path != wantPath || op.Action != string(tapc.RecordCreateAction) {
				continue
			}
			announced = true
			assert.NotNil(t, op.Cid, "the live op cites the record cid")
			assert.NotEmpty(t, frame.Commit.Blocks, "the live commit frame includes its CAR")
		}
	}
	require.True(t, announced, "a live commit frame announced the seeded record")

	pushToken := os.Getenv("KNOT_TEST_PUSH_TOKEN")
	require.NotEmpty(t, pushToken, "KNOT_TEST_PUSH_TOKEN authorizes the receive-pack push")
	pushRepo(t, source, addr, seeded.Did, pushToken)
	records, cursor := knotfeedClient(t, addr, did)
	sawRecord := false
	sawRefs := false
	deadline := time.After(60 * time.Second)
	var settle <-chan time.Time
replay:
	for {
		if sawRecord && sawRefs && settle == nil {
			settle = time.After(2 * time.Second)
		}
		select {
		case op := <-records:
			assert.NotEqual(t, tapc.RecordDeleteAction, op.Action,
				"replay from zero mustn't serve a delete for %s/%s", op.Collection.String(), op.Rkey.String())
			switch {
			case op.Collection.String() == ticketCollection && op.Rkey == rkey:
				assert.Equal(t, tapc.RecordCreateAction, op.Action, "the replayed seed op is a create")
				assert.NotEmpty(t, op.Bytes, "the replayed op includes the record bytes")
				sawRecord = true
			case op.Collection.String() == "org.tangled.git.refs":
				assert.Equal(t, "self", op.Rkey.String(), "the refs snapshot is a single self record")
				def, refs, err := knotfeed.DecodeRefsSnapshot(op.Bytes)
				require.NoError(t, err, "the frozen record envelope decodes")
				if def != "main" {
					break
				}
				for _, line := range strings.Split(string(refs), "\n") {
					line = strings.TrimSpace(line)
					sha, name, ok := strings.Cut(line, " ")
					if !ok || name != "refs/heads/main" || sha != wantSha {
						continue
					}
					sawRefs = true
				}
			}
		case <-settle:
			break replay
		case <-deadline:
			t.Fatalf("replay served the seeded record %v and the refs snapshot %v", sawRecord, sawRefs)
		}
	}
	require.Eventually(t, func() bool { return cursor.Load() >= int64(1) },
		5*time.Second, 10*time.Millisecond, "the consumer advanced its cursor")
}