package hydrant

import (
	"bufio"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net"
	"net/http"
	"os"
	"path/filepath"
	"strings"
	"testing"
	"time"
)

// settings for a hydrant that never touches the network, so the test runs offline
func offlineSettings(dir string) map[string]string {
	return map[string]string{
		"HYDRANT_DATABASE_PATH":  filepath.Join(dir, "db"),
		"HYDRANT_API_BIND":       "unix:" + filepath.Join(dir, "api.sock"),
		"HYDRANT_RELAY_HOSTS":    "",
		"HYDRANT_SEED_HOSTS":     "",
		"HYDRANT_CRAWLER_URLS":   "",
		"HYDRANT_ENABLE_CRAWLER": "false",
		"RUST_LOG":               "warn",
	}
}

// a short temp dir, because unix socket paths are capped at ~104 bytes on darwin
func shortTempDir(t *testing.T) string {
	t.Helper()
	dir, err := os.MkdirTemp("/tmp", "hydrant-ffi-")
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { os.RemoveAll(dir) })
	return dir
}

// starts hydrant and shuts it down when the test ends, before its dir is removed
func startHydrant(t *testing.T, settings map[string]string) *Hydrant {
	t.Helper()
	h, err := Start(settings)
	if err != nil {
		t.Fatalf("start: %v", err)
	}
	t.Cleanup(func() {
		if err := h.Shutdown(10 * time.Second); err != nil {
			t.Errorf("shutdown: %v", err)
		}
	})
	return h
}

func unixClient(socket string) *http.Client {
	return &http.Client{
		Timeout: 5 * time.Second,
		Transport: &http.Transport{
			DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
				var d net.Dialer
				return d.DialContext(ctx, "unix", socket)
			},
		},
	}
}

func waitReady(t *testing.T, h *Hydrant, client *http.Client) {
	t.Helper()
	deadline := time.Now().Add(30 * time.Second)
	for time.Now().Before(deadline) {
		select {
		case <-h.Done():
			t.Fatalf("hydrant stopped before its api came up: %v", h.Err())
		default:
		}
		if resp, err := client.Get("http://hydrant/stats"); err == nil {
			resp.Body.Close()
			if resp.StatusCode == http.StatusOK {
				return
			}
		}
		time.Sleep(100 * time.Millisecond)
	}
	t.Fatal("hydrant api never answered on the socket")
}

func request(t *testing.T, client *http.Client, method, path, body string) string {
	t.Helper()
	req, err := http.NewRequest(method, "http://hydrant"+path, strings.NewReader(body))
	if err != nil {
		t.Fatal(err)
	}
	if body != "" {
		req.Header.Set("Content-Type", "application/json")
	}
	resp, err := client.Do(req)
	if err != nil {
		t.Fatalf("%s %s: %v", method, path, err)
	}
	defer resp.Body.Close()
	out, _ := io.ReadAll(resp.Body)
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("%s %s: status %d: %s", method, path, resp.StatusCode, out)
	}
	return string(out)
}

// opens a /stream websocket and leaves it open until the test ends
func subscribe(t *testing.T, socket string) net.Conn {
	t.Helper()
	conn, err := net.Dial("unix", socket)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { conn.Close() })
	fmt.Fprint(conn, "GET /stream HTTP/1.1\r\nHost: hydrant\r\nConnection: Upgrade\r\nUpgrade: websocket\r\n"+
		"Sec-WebSocket-Version: 13\r\nSec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n\r\n")
	conn.SetReadDeadline(time.Now().Add(5 * time.Second))
	status, err := bufio.NewReader(conn).ReadString('\n')
	if err != nil {
		t.Fatalf("read upgrade response: %v", err)
	}
	if !strings.Contains(status, "101") {
		t.Errorf("/stream didn't upgrade over the socket: %q", status)
	}
	return conn
}

func TestEmbeddedHydrantServesOverUnixSocket(t *testing.T) {
	dir := shortTempDir(t)
	settings := offlineSettings(dir)
	socket := strings.TrimPrefix(settings["HYDRANT_API_BIND"], "unix:")

	// the embedded config must come only from the settings map
	envDB := filepath.Join(dir, "from-env")
	t.Setenv("HYDRANT_DATABASE_PATH", envDB)

	h := startHydrant(t, settings)
	client := unixClient(socket)
	waitReady(t, h, client)

	if _, err := os.Stat(settings["HYDRANT_DATABASE_PATH"]); err != nil {
		t.Errorf("database not created at the configured path: %v", err)
	}
	if _, err := os.Stat(envDB); err == nil {
		t.Errorf("hydrant read HYDRANT_DATABASE_PATH from the process environment")
	}

	info, err := os.Stat(socket)
	if err != nil {
		t.Fatal(err)
	}
	if mode := info.Mode().Perm(); mode != 0o600 {
		t.Errorf("socket mode is %o, want 600", mode)
	}

	request(t, client, http.MethodPatch, "/filter", `{"mode":"filter","signals":["sh.tangled.repo"]}`)
	var filter struct {
		Mode    string   `json:"mode"`
		Signals []string `json:"signals"`
	}
	if err := json.Unmarshal([]byte(request(t, client, http.MethodGet, "/filter", "")), &filter); err != nil {
		t.Fatal(err)
	}
	if filter.Mode != "filter" || len(filter.Signals) != 1 || filter.Signals[0] != "sh.tangled.repo" {
		t.Errorf("filter didn't stick: %+v", filter)
	}

	request(t, client, http.MethodGet, "/stream/head", "")

	subscribe(t, socket)

	t.Run("a second hydrant on the same database is refused", func(t *testing.T) {
		again := offlineSettings(dir)
		again["HYDRANT_API_BIND"] = "unix:" + filepath.Join(dir, "other.sock")
		_, err := Start(again)
		if err == nil || !strings.Contains(err.Error(), "Locked") {
			t.Fatalf("got %v, want the database lock to refuse it", err)
		}
	})

	t.Run("a socket that is still served is not taken over", func(t *testing.T) {
		other := offlineSettings(shortTempDir(t))
		other["HYDRANT_API_BIND"] = settings["HYDRANT_API_BIND"]
		h2 := startHydrant(t, other)
		select {
		case <-h2.Done():
			if !strings.Contains(h2.Err().Error(), "listening on this socket") {
				t.Errorf("unexpected error: %v", h2.Err())
			}
		case <-time.After(30 * time.Second):
			t.Fatal("second hydrant kept running on a socket that was in use")
		}
		// the first one must still be serving
		request(t, client, http.MethodGet, "/stats", "")
	})
}

func TestShutdownLetsTheSameDatabaseStartAgain(t *testing.T) {
	dir := shortTempDir(t)
	settings := offlineSettings(dir)
	socket := strings.TrimPrefix(settings["HYDRANT_API_BIND"], "unix:")
	client := unixClient(socket)

	h, err := Start(settings)
	if err != nil {
		t.Fatalf("start: %v", err)
	}
	waitReady(t, h, client)
	request(t, client, http.MethodPatch, "/filter", `{"mode":"filter","signals":["sh.tangled.repo"]}`)
	// a subscriber holds a stream thread, which has to end too
	subscribe(t, socket)

	if err := h.Shutdown(10 * time.Second); err != nil {
		t.Fatalf("shutdown: %v", err)
	}
	if !errors.Is(h.Err(), ErrShutdown) {
		t.Errorf("Err after Shutdown is %v, want ErrShutdown", h.Err())
	}
	if err := h.Shutdown(time.Second); err == nil {
		t.Error("a second Shutdown should be refused")
	}

	// the database lock refuses this unless the first hydrant really let go
	again := startHydrant(t, settings)
	waitReady(t, again, client)
	if filter := request(t, client, http.MethodGet, "/filter", ""); !strings.Contains(filter, "sh.tangled.repo") {
		t.Errorf("the filter from before the shutdown is gone: %s", filter)
	}
}

func TestShutdownAfterTheSubscriberHungUp(t *testing.T) {
	dir := shortTempDir(t)
	settings := offlineSettings(dir)
	socket := strings.TrimPrefix(settings["HYDRANT_API_BIND"], "unix:")
	h, err := Start(settings)
	if err != nil {
		t.Fatalf("start: %v", err)
	}
	waitReady(t, h, unixClient(socket))

	// hydrant's goodbye to this socket is a write into a closed pipe, and on
	// linux that raises SIGPIPE on a rust thread, which go doesn't handle. the
	// sleep lets it get there before the shutdown cuts that task off
	subscribe(t, socket).Close()
	time.Sleep(200 * time.Millisecond)
	if err := h.Shutdown(10 * time.Second); err != nil {
		t.Fatalf("shutdown: %v", err)
	}
}

func TestSocketModeIsConfigurable(t *testing.T) {
	dir := shortTempDir(t)
	settings := offlineSettings(dir)
	settings["HYDRANT_API_SOCKET_MODE"] = "0660"
	socket := strings.TrimPrefix(settings["HYDRANT_API_BIND"], "unix:")
	h := startHydrant(t, settings)
	waitReady(t, h, unixClient(socket))

	info, err := os.Stat(socket)
	if err != nil {
		t.Fatal(err)
	}
	if mode := info.Mode().Perm(); mode != 0o660 {
		t.Fatalf("socket mode is %o, want 660", mode)
	}
}

func TestNoApiKeepsRunning(t *testing.T) {
	dir := shortTempDir(t)
	settings := offlineSettings(dir)
	settings["HYDRANT_API_BIND"] = "none"
	h := startHydrant(t, settings)
	// with nothing to serve, a serve future that returned right away would end hydrant here
	select {
	case <-h.Done():
		t.Fatalf("hydrant stopped without an api: %v", h.Err())
	case <-time.After(2 * time.Second):
	}
	entries, err := os.ReadDir(dir)
	if err != nil {
		t.Fatal(err)
	}
	for _, e := range entries {
		if e.Type()&os.ModeSocket != 0 {
			t.Fatalf("no api should bind nothing, found socket %s", e.Name())
		}
	}
}

func TestStartRejectsBadSettings(t *testing.T) {
	for name, tc := range map[string]struct {
		mutate func(map[string]string)
		want   string
	}{
		"missing bind": {func(s map[string]string) { delete(s, "HYDRANT_API_BIND") }, "HYDRANT_API_BIND is required"},
		"bad bind":     {func(s map[string]string) { s["HYDRANT_API_BIND"] = "nope" }, "invalid HYDRANT_API_BIND"},
		"bad setting":  {func(s map[string]string) { s["HYDRANT_HISTORY_TTL"] = "soon" }, "HYDRANT_HISTORY_TTL"},
		"bad mode":     {func(s map[string]string) { s["HYDRANT_API_SOCKET_MODE"] = "0800" }, "HYDRANT_API_SOCKET_MODE"},
	} {
		t.Run(name, func(t *testing.T) {
			settings := offlineSettings(shortTempDir(t))
			tc.mutate(settings)
			_, err := Start(settings)
			if err == nil || !strings.Contains(err.Error(), tc.want) {
				t.Fatalf("got %v, want an error mentioning %q", err, tc.want)
			}
		})
	}
}