package bell_connector_test

import (
	"bytes"
	"context"
	"crypto/ed25519"
	"crypto/rand"
	"crypto/tls"
	"encoding/json"
	"net/http"
	"net/http/httptest"
	"os"
	"path/filepath"
	"strings"
	"sync/atomic"
	"testing"
	"time"

	"github.com/gin-gonic/gin"
	"gorm.io/driver/sqlite"
	"gorm.io/gorm"

	"git.ilapage.cn/ila/yovision/Sense/server/app/sense/integration/bell_connector"
	"git.ilapage.cn/ila/yovision/Sense/server/app/sense/integration/machine_identity"
	"git.ilapage.cn/ila/yovision/Sense/server/app/sense/outbox"
)

func TestPersistentOutboxRetriesWithoutChangingOriginalIdentity(t *testing.T) {
	db, databasePath := openDatabase(t)
	body := fixture(t)
	_, privateKey, _ := ed25519.GenerateKey(rand.Reader)
	signer := machine_identity.Signer{Principal: "yv:sense:school-a", KeyID: "sense-key-0001", PrivateKey: privateKey, Now: func() time.Time { return time.Date(2026, 8, 31, 1, 0, 0, 0, time.UTC) }}
	var attempts atomic.Int32
	var received [][]byte
	server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
		payload := make([]byte, request.ContentLength)
		_, _ = request.Body.Read(payload)
		received = append(received, payload)
		requestID := request.Header.Get("X-Request-ID")
		writer.Header().Set("X-Request-ID", requestID)
		if request.Header.Get("X-YoVision-Relay-ID") != "sense-school-a" || !strings.HasPrefix(request.Header.Get("Authorization"), "Bearer ") || len(requestID) < 16 || len(requestID) > 128 {
			http.Error(writer, "missing relay identity", http.StatusUnauthorized)
			return
		}
		if attempts.Add(1) == 1 {
			writer.WriteHeader(http.StatusServiceUnavailable)
			_, _ = writer.Write([]byte(`{"code":"ingest_unavailable","message":"temporarily unavailable"}`))
			return
		}
		writer.Header().Set("Content-Type", "application/json")
		writer.WriteHeader(http.StatusCreated)
		_, _ = writer.Write([]byte(`{"event_id":"bell-event-1","producer_id":"brain-school-a","source_event_id":"evt-area-20260831-0001","disposition":"created","payload_sha256":"4cc1e93820195caf713ea675ff33f178c9d4997dd8a81cb61287e9fea0e3d5e1"}`))
	}))
	defer server.Close()

	now := time.Date(2026, 8, 31, 1, 0, 0, 0, time.UTC)
	if _, err := bell_connector.EnqueueEvent(db, body, now); err != nil {
		t.Fatal(err)
	}
	client := &bell_connector.Client{Endpoint: server.URL, RelayID: "sense-school-a", Signer: signer, HTTP: server.Client(), Enabled: true}
	relay := bell_connector.Relay{DB: db, Client: client, Now: func() time.Time { return now }, Backoff: func(int) time.Duration { return 0 }}
	if delivered, err := relay.DeliverBatch(context.Background(), "worker-1", 10); err != nil || delivered != 0 {
		t.Fatalf("unavailable delivery=%d err=%v", delivered, err)
	}
	var queued outbox.Message
	if err := db.First(&queued).Error; err != nil || queued.State != outbox.StateRetry {
		t.Fatalf("outbox was not retained for retry: state=%s err=%v", queued.State, err)
	}
	sqlDatabase, err := db.DB()
	if err != nil {
		t.Fatal(err)
	}
	if err = sqlDatabase.Close(); err != nil {
		t.Fatal(err)
	}
	db = openDatabasePath(t, databasePath)
	if err = db.First(&queued).Error; err != nil {
		t.Fatal(err)
	}

	// Constructing a new relay simulates restart; the durable row is claimed
	// and delivered with its original bytes and business identity.
	restarted := bell_connector.Relay{DB: db, Client: client, Now: func() time.Time { return now }, Backoff: func(int) time.Duration { return 0 }}
	if delivered, err := restarted.DeliverBatch(context.Background(), "worker-2", 10); err != nil || delivered != 1 {
		t.Fatalf("recovery delivery=%d err=%v", delivered, err)
	}
	if len(received) != 2 || string(received[0]) != string(body) || string(received[1]) != string(body) || bell_connector.PreserveIdentity(received[0], received[1]) != nil {
		t.Fatal("relay changed the frozen payload or original identity")
	}
	if err := db.First(&queued).Error; err != nil || queued.State != outbox.StateDelivered {
		t.Fatalf("outbox not delivered: state=%s err=%v", queued.State, err)
	}
}

func TestTerminalConflictDisabledTimeoutAndReplayRestart(t *testing.T) {
	db, databasePath := openDatabase(t)
	body := fixture(t)
	_, privateKey, _ := ed25519.GenerateKey(rand.Reader)
	signer := machine_identity.Signer{Principal: "yv:sense:school-a", KeyID: "sense-key-0001", PrivateKey: privateKey}
	policy := machine_identity.TransportPolicy{TLSMinVersion: tls.VersionTLS12, VerifyCertificate: true, VerifyHostname: true, ConnectTimeout: time.Second, ResponseHeaderTimeout: time.Second, RequestTimeout: time.Second, MaxRequestBytes: 64 * 1024}
	if _, err := bell_connector.NewClient("https://user:pass@bell.example", "sense-school-a", signer, policy); err == nil || !strings.Contains(err.Error(), "HTTPS origin") {
		t.Fatalf("endpoint userinfo was not rejected: %v", err)
	}
	conflictServer := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
		writer.WriteHeader(http.StatusConflict)
		_, _ = writer.Write([]byte(`{"code":"idempotency_conflict","message":"conflict","existing_event_id":"bell-1"}`))
	}))
	defer conflictServer.Close()
	if _, err := bell_connector.EnqueueEvent(db, body, time.Now()); err != nil {
		t.Fatal(err)
	}
	client := &bell_connector.Client{Endpoint: conflictServer.URL, Signer: signer, HTTP: conflictServer.Client(), Enabled: true}
	if _, err := (bell_connector.Relay{DB: db, Client: client}).DeliverBatch(context.Background(), "worker", 1); err != nil {
		t.Fatal(err)
	}
	var message outbox.Message
	if err := db.First(&message).Error; err != nil || message.State != outbox.StateDead {
		t.Fatalf("terminal conflict was retried: state=%s err=%v", message.State, err)
	}
	if _, err := (bell_connector.Client{Enabled: false}).Send(context.Background(), body); err == nil || !strings.Contains(err.Error(), "connector_disabled") {
		t.Fatalf("disabled connector error=%v", err)
	}
	cancelled, cancel := context.WithCancel(context.Background())
	cancel()
	if _, err := client.Send(cancelled, body); err == nil {
		t.Fatal("cancelled/timeout request unexpectedly succeeded")
	}

	now := time.Now().UTC()
	firstStore := bell_connector.PersistentReplayStore{DB: db}
	if !firstStore.Consume("yv:bell:school-a", "abcdefghijklmnopqrstuv", now.Add(time.Minute), now) {
		t.Fatal("first replay consume failed")
	}
	sqlDatabase, err := db.DB()
	if err != nil {
		t.Fatal(err)
	}
	if err = sqlDatabase.Close(); err != nil {
		t.Fatal(err)
	}
	db = openDatabasePath(t, databasePath)
	restartedStore := bell_connector.PersistentReplayStore{DB: db}
	if restartedStore.Consume("yv:bell:school-a", "abcdefghijklmnopqrstuv", now.Add(time.Minute), now) {
		t.Fatal("replay was accepted after store restart")
	}
}

func TestBrainIngressAndBellEvidenceEndpoint(t *testing.T) {
	gin.SetMode(gin.TestMode)
	db, _ := openDatabase(t)
	brainPublic, brainPrivate, _ := ed25519.GenerateKey(rand.Reader)
	bellPublic, bellPrivate, _ := ed25519.GenerateKey(rand.Reader)
	registry, err := machine_identity.NewRegistry(
		machine_identity.KeyRecord{Principal: "yv:brain:school-a", KeyID: "brain-key-0001", PublicKey: brainPublic, Audience: "yovision-sense", Scopes: []string{"events:ingest"}, Enabled: true},
		machine_identity.KeyRecord{Principal: "yv:bell:school-a", KeyID: "bell-key-0001", PublicKey: bellPublic, Audience: "yovision-sense", Scopes: []string{"evidence:read"}, Enabled: true},
	)
	if err != nil {
		t.Fatal(err)
	}
	now := time.Date(2026, 8, 31, 1, 0, 0, 0, time.UTC)
	verifier := machine_identity.Verifier{Registry: registry, Replay: bell_connector.PersistentReplayStore{DB: db}, Now: func() time.Time { return now }}
	body := fixture(t)
	brainSigner := machine_identity.Signer{Principal: "yv:brain:school-a", KeyID: "brain-key-0001", PrivateKey: brainPrivate, Now: func() time.Time { return now }}
	token, err := brainSigner.Mint("yovision-sense", []string{"events:ingest"}, http.MethodPost, "/v1/events", body)
	if err != nil {
		t.Fatal(err)
	}
	request := httptest.NewRequest(http.MethodPost, "/v1/events", bytes.NewReader(body))
	request.Header.Set("Authorization", "Bearer "+token)
	request.Header.Set("X-Request-ID", "request-id-0000001")
	response := httptest.NewRecorder()
	ginContext, _ := gin.CreateTestContext(response)
	ginContext.Request = request
	(bell_connector.IngressHandler{DB: db, Verifier: verifier, EvidenceOwnerID: "sense-school-a", Now: func() time.Time { return now }}).Post(ginContext)
	if response.Code != http.StatusAccepted {
		t.Fatalf("Brain ingress status=%d body=%s", response.Code, response.Body.String())
	}
	var evidence bell_connector.EvidenceRecord
	if err = db.First(&evidence, "evidence_id = ?", "ev-school-east-0001").Error; err != nil {
		t.Fatal(err)
	}
	bellSigner := machine_identity.Signer{Principal: "yv:bell:school-a", KeyID: "bell-key-0001", PrivateKey: bellPrivate, Now: func() time.Time { return now }}
	path := "/v1/evidence/ev-school-east-0001"
	token, err = bellSigner.Mint("yovision-sense", []string{"evidence:read"}, http.MethodGet, path, nil)
	if err != nil {
		t.Fatal(err)
	}
	getRequest := httptest.NewRequest(http.MethodGet, path, nil)
	getRequest.Header.Set("Authorization", "Bearer "+token)
	getRequest.Header.Set("X-Request-ID", "request-id-0000002")
	getResponse := httptest.NewRecorder()
	getContext, _ := gin.CreateTestContext(getResponse)
	getContext.Request = getRequest
	getContext.Params = gin.Params{{Key: "evidence_id", Value: "ev-school-east-0001"}}
	(bell_connector.EvidenceHandler{DB: db, Verifier: verifier, Now: func() time.Time { return now }}).Get(getContext)
	if getResponse.Code != http.StatusOK || !bytes.Equal(getResponse.Body.Bytes(), evidence.Payload) {
		t.Fatalf("evidence lookup status=%d body=%s", getResponse.Code, getResponse.Body.String())
	}
}

func openDatabase(t *testing.T) (*gorm.DB, string) {
	databasePath := filepath.Join(t.TempDir(), "sense-bell-connector.sqlite")
	return openDatabasePath(t, databasePath), databasePath
}

func openDatabasePath(t *testing.T, databasePath string) *gorm.DB {
	t.Helper()
	db, err := gorm.Open(sqlite.Open(databasePath), &gorm.Config{})
	if err != nil {
		t.Fatal(err)
	}
	if err = db.AutoMigrate(&outbox.Message{}, &outbox.DeliveryRecord{}, &outbox.Attempt{}, &bell_connector.ReplayToken{}, &bell_connector.InboundEvent{}, &bell_connector.EvidenceRecord{}); err != nil {
		t.Fatal(err)
	}
	sqlDatabase, err := db.DB()
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = sqlDatabase.Close() })
	return db
}

func fixture(t *testing.T) []byte {
	t.Helper()
	path := filepath.Join("..", "..", "..", "..", "contracts", "events", "v1", "examples", "dangerous-area.json")
	body, err := os.ReadFile(path)
	if err != nil {
		t.Fatal(err)
	}
	var event map[string]any
	if json.Unmarshal(body, &event) != nil {
		t.Fatal("invalid fixture")
	}
	return body
}