package app

import (
	"bufio"
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"net/http/httptest"
	"net/url"
	"os"
	"strings"
	"sync/atomic"
	"testing"
	"time"

	"github.com/alicebob/miniredis/v2"
	miniredisserver "github.com/alicebob/miniredis/v2/server"
	"github.com/gorilla/websocket"
	"github.com/stretchr/testify/require"
	"gorm.io/driver/postgres"
	"gorm.io/gorm"
	"gorm.io/gorm/logger"

	handlerws "github.com/gochat/gochat/internal/handler/ws"
	"github.com/gochat/gochat/internal/model"
	wspkg "github.com/gochat/gochat/internal/ws"
)

func TestBootstrapRoutesDurableRealtimeToHubAndSSEWithoutDuplicates(t *testing.T) {
	if os.Getenv("GOCHAT_TEST_DB") == "sqlite" {
		t.Skip("bootstrap production wiring requires PostgreSQL")
	}
	dsn := os.Getenv("GOCHAT_TEST_DB_URL")
	if dsn == "" {
		t.Skip("GOCHAT_TEST_DB_URL is not set")
	}

	db, err := gorm.Open(postgres.Open(dsn), &gorm.Config{Logger: logger.Default.LogMode(logger.Silent)})
	require.NoError(t, err)
	sqlDB, err := db.DB()
	require.NoError(t, err)
	require.NoError(t, sqlDB.Ping())
	t.Cleanup(func() { require.NoError(t, sqlDB.Close()) })
	schema := fmt.Sprintf("hh564_bootstrap_%d", time.Now().UnixNano())
	require.NoError(t, db.Exec(fmt.Sprintf(`CREATE SCHEMA %q`, schema)).Error)
	t.Cleanup(func() { require.NoError(t, db.Exec(fmt.Sprintf(`DROP SCHEMA %q CASCADE`, schema)).Error) })
	databaseURL, err := url.Parse(dsn)
	require.NoError(t, err)
	query := databaseURL.Query()
	query.Set("search_path", schema+",public")
	databaseURL.RawQuery = query.Encode()

	mini := miniredis.RunT(t)
	t.Setenv("GOCHAT_DATABASE_DSN", databaseURL.String())
	t.Setenv("GOCHAT_DATABASE_RUN_MIGRATIONS", "true")
	t.Setenv("GOCHAT_REDIS_DSN", "redis://"+mini.Addr()+"/0")
	t.Setenv("GOCHAT_JWT_ALLOW_INSECURE_HEADER_AUTH", "true")
	t.Setenv("GOCHAT_SEARCH_ENGINE", "db")
	t.Setenv("GOCHAT_RATE_LIMIT_ENABLED", "false")
	t.Setenv("GOCHAT_LOG_LEVEL", "error")
	t.Chdir("../..")

	application, err := Bootstrap("default")
	require.NoError(t, err)
	relayCtx, stopRelay := context.WithCancel(context.Background())
	require.NoError(t, application.wsRelay.Start(relayCtx))
	t.Cleanup(func() {
		stopRelay()
		if err := application.Shutdown(2 * time.Second); err != nil && !strings.Contains(err.Error(), "redis: client is closed") {
			t.Errorf("shutdown bootstrap app: %v", err)
		}
	})

	require.NoError(t, application.db.Exec("DELETE FROM background_jobs").Error)
	stamp := time.Now().UnixNano()
	account := model.Account{Name: fmt.Sprintf("bootstrap-sse-%d", stamp), Active: true}
	require.NoError(t, application.db.Create(&account).Error)
	user := model.User{AccountID: account.ID, Name: "Bootstrap SSE", Email: fmt.Sprintf("bootstrap-sse-%d@example.test", stamp), Password: "unused", Provider: "email", Active: true}
	require.NoError(t, application.db.Create(&user).Error)
	require.NoError(t, application.db.Create(&model.AccountUser{UserID: user.ID, AccountID: account.ID, Role: "administrator"}).Error)
	visitorToken := fmt.Sprintf("bootstrap-visitor-%d", stamp)
	inbox := model.Inbox{AccountID: account.ID, Name: "Bootstrap widget", ChannelType: "Channel::WebWidget", Enabled: true}
	require.NoError(t, application.db.Create(&inbox).Error)
	contact := model.Contact{AccountID: account.ID, Name: "Bootstrap visitor"}
	require.NoError(t, application.db.Create(&contact).Error)
	require.NoError(t, application.db.Create(&model.ContactInbox{ContactID: contact.ID, InboxID: inbox.ID, PubsubToken: visitorToken}).Error)

	dashboard := handlerws.NewClient(user.ID, account.ID, nil, application.wsHub)
	dashboard.Identifier = fmt.Sprintf(`{"channel":"AccountChannel","account_id":%d}`, account.ID)
	application.wsHub.Register(dashboard)
	visitor := handlerws.NewClient(0, account.ID, nil, application.wsHub)
	visitor.IsContact = true
	visitor.PubsubToken = visitorToken
	visitor.Identifier = fmt.Sprintf(`{"channel":"RoomChannel","pubsub_token":%q}`, visitorToken)
	application.wsHub.Register(visitor)
	t.Cleanup(func() {
		application.wsHub.Unregister(dashboard)
		application.wsHub.Unregister(visitor)
	})

	server := httptest.NewServer(application.Handler())
	t.Cleanup(server.Close)
	visitorConn, _, err := websocket.DefaultDialer.Dial(
		"ws"+strings.TrimPrefix(server.URL, "http")+"/cable?pubsub_token="+url.QueryEscape(visitorToken),
		nil,
	)
	require.NoError(t, err)
	t.Cleanup(func() { require.NoError(t, visitorConn.Close()) })
	require.NoError(t, visitorConn.SetReadDeadline(time.Now().Add(time.Second)))
	_, _, err = visitorConn.ReadMessage()
	require.NoError(t, err)
	identifier := fmt.Sprintf(`{"channel":"RoomChannel","pubsub_token":%q}`, visitorToken)
	require.NoError(t, visitorConn.WriteJSON(map[string]any{"command": "subscribe", "identifier": identifier}))
	var confirmation map[string]any
	require.NoError(t, visitorConn.ReadJSON(&confirmation))
	require.Equal(t, "confirm_subscription", confirmation["type"])
	require.JSONEq(t, identifier, confirmation["identifier"].(string))

	requestCtx, cancelRequest := context.WithCancel(context.Background())
	t.Cleanup(cancelRequest)
	req, err := http.NewRequestWithContext(requestCtx, http.MethodGet, fmt.Sprintf("%s/api/v1/accounts/%d/events", server.URL, account.ID), nil)
	require.NoError(t, err)
	req.Header.Set("X-User-ID", fmt.Sprint(user.ID))
	req.Header.Set("X-Account-ID", fmt.Sprint(account.ID))
	resp, err := server.Client().Do(req)
	require.NoError(t, err)
	t.Cleanup(func() { require.NoError(t, resp.Body.Close()) })
	require.Equal(t, http.StatusOK, resp.StatusCode)
	require.Equal(t, "text/event-stream", resp.Header.Get("Content-Type"))
	sse := bufio.NewReader(resp.Body)
	eventType, _ := readBootstrapSSE(t, sse)
	require.Equal(t, "connected", eventType)

	tokenChannel := wspkg.RedisPrefixRoom + "pubsub_token_" + visitorToken
	var tokenFailed atomic.Bool
	mini.Server().SetPreHook(func(peer *miniredisserver.Peer, command string, args ...string) bool {
		if strings.EqualFold(command, "publish") && len(args) > 0 && args[0] == tokenChannel && tokenFailed.CompareAndSwap(false, true) {
			peer.WriteError("token room unavailable")
			return true
		}
		return false
	})

	require.NoError(t, application.eventPublisher.PublishWidgetEvent(account.ID, visitorToken, wspkg.EventMessageCreated, map[string]any{"id": 7}))
	processed, err := application.workerPool.ProcessOne(context.Background())
	require.True(t, processed)
	require.NoError(t, err)
	requireHubEvent(t, dashboard.Send)
	eventType, payload := readBootstrapSSE(t, sse)
	require.Equal(t, wspkg.EventMessageCreated, eventType)
	require.JSONEq(t, `{"id":7}`, payload)

	processed, err = application.workerPool.ProcessOne(context.Background())
	require.True(t, processed)
	require.ErrorContains(t, err, "token room unavailable")
	select {
	case <-visitor.Send:
		t.Fatal("token subscriber received failed delivery")
	default:
	}
	require.NoError(t, application.db.Model(&model.BackgroundJob{}).
		Where("status = ?", model.BackgroundJobStatusRetrying).
		Update("scheduled_at", time.Now().Add(-time.Second)).Error)

	processed, err = application.workerPool.ProcessOne(context.Background())
	require.True(t, processed)
	require.NoError(t, err)
	requireHubEvent(t, visitor.Send)
	var delivered struct {
		Identifier string          `json:"identifier"`
		Message    json.RawMessage `json:"message"`
	}
	require.NoError(t, visitorConn.ReadJSON(&delivered))
	require.JSONEq(t, identifier, delivered.Identifier)
	var widgetEvent wspkg.WSMessage
	require.NoError(t, json.Unmarshal(delivered.Message, &widgetEvent))
	require.Equal(t, wspkg.EventMessageCreated, widgetEvent.Event)
	require.Equal(t, float64(7), widgetEvent.Data.(map[string]any)["id"])
	require.True(t, tokenFailed.Load())
	assertNoHubEvent(t, dashboard.Send)
	assertNoHubEvent(t, visitor.Send)
	assertNoBootstrapSSE(t, sse)
}

func readBootstrapSSE(t *testing.T, reader *bufio.Reader) (string, string) {
	t.Helper()
	resultCh := make(chan bootstrapSSEResult, 1)
	go func() {
		resultCh <- scanBootstrapSSE(reader)
	}()
	select {
	case got := <-resultCh:
		require.NoError(t, got.err)
		return got.event, got.data
	case <-time.After(time.Second):
		t.Fatal("timed out waiting for SSE event")
		return "", ""
	}
}

func requireHubEvent(t *testing.T, events <-chan []byte) {
	t.Helper()
	select {
	case <-events:
	case <-time.After(time.Second):
		t.Fatal("timed out waiting for Hub event")
	}
}

func assertNoHubEvent(t *testing.T, events <-chan []byte) {
	t.Helper()
	time.Sleep(20 * time.Millisecond)
	select {
	case <-events:
		t.Fatal("subscriber received duplicate event")
	default:
	}
}

func assertNoBootstrapSSE(t *testing.T, reader *bufio.Reader) {
	t.Helper()
	eventCh := make(chan bootstrapSSEResult, 1)
	go func() {
		eventCh <- scanBootstrapSSE(reader)
	}()
	select {
	case event := <-eventCh:
		require.NoError(t, event.err)
		t.Fatalf("SSE subscriber received duplicate %s", event.event)
	case <-time.After(50 * time.Millisecond):
	}
}

type bootstrapSSEResult struct {
	event string
	data  string
	err   error
}

func scanBootstrapSSE(reader *bufio.Reader) bootstrapSSEResult {
	var got bootstrapSSEResult
	for {
		line, err := reader.ReadString('\n')
		if err != nil {
			got.err = err
			return got
		}
		line = strings.TrimSpace(line)
		switch {
		case strings.HasPrefix(line, "event: "):
			got.event = strings.TrimPrefix(line, "event: ")
		case strings.HasPrefix(line, "data: "):
			got.data = strings.TrimPrefix(line, "data: ")
		case line == "":
			return got
		}
	}
}