package edge

import (
	"bytes"
	"context"
	"encoding/json"
	"net"
	"net/http"
	"os"
	"path/filepath"
	"testing"
	"time"
)

func dialPurgeSocket(t *testing.T, socketPath string) *http.Client {
	t.Helper()
	return &http.Client{
		Transport: &http.Transport{
			DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
				return net.Dial("unix", socketPath)
			},
		},
	}
}

func TestListenPurgeRemovesCachedEntry(t *testing.T) {
	dir := t.TempDir()
	sock := filepath.Join(dir, "instance.sock")
	cache := NewCache(DefaultCacheMaxEntries, DefaultCacheMaxBodyBytes)
	cache.Put(CacheKey{Host: "a.example", Method: "GET", Path: "/event"}, "", &CachedResponse{
		StatusCode: 200, Header: http.Header{}, Body: []byte("hi"), StoredAt: time.Now(), TTL: time.Minute,
	})

	pl, err := ListenPurge(sock, "a.example", cache)
	if err != nil {
		t.Fatalf("ListenPurge: %v", err)
	}
	defer pl.Close()

	body, _ := json.Marshal(map[string]string{"path": "/event", "query": ""})
	client := dialPurgeSocket(t, sock)
	resp, err := client.Post("http://unix/purge", "application/json", bytes.NewReader(body))
	if err != nil {
		t.Fatalf("purge POST: %v", err)
	}
	resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("purge status = %d, want 200", resp.StatusCode)
	}

	if _, ok := cache.get(CacheKey{Host: "a.example", Method: "GET", Path: "/event"}); ok {
		t.Fatal("expected entry to be purged")
	}
}

func TestListenPurgeCannotAffectAnotherHost(t *testing.T) {
	dir := t.TempDir()
	sock := filepath.Join(dir, "instance.sock")
	cache := NewCache(DefaultCacheMaxEntries, DefaultCacheMaxBodyBytes)
	cache.Put(CacheKey{Host: "other.example", Method: "GET", Path: "/event"}, "", &CachedResponse{
		StatusCode: 200, Header: http.Header{}, Body: []byte("hi"), StoredAt: time.Now(), TTL: time.Minute,
	})

	pl, err := ListenPurge(sock, "a.example", cache) // bound to a.example, not other.example
	if err != nil {
		t.Fatalf("ListenPurge: %v", err)
	}
	defer pl.Close()

	body, _ := json.Marshal(map[string]string{"path": "/event", "query": ""})
	client := dialPurgeSocket(t, sock)
	resp, _ := client.Post("http://unix/purge", "application/json", bytes.NewReader(body))
	resp.Body.Close()

	if _, ok := cache.get(CacheKey{Host: "other.example", Method: "GET", Path: "/event"}); !ok {
		t.Fatal("a purge listener bound to a.example must never affect other.example's cache entries")
	}
}

func TestListenPurgeSocketFileIsOwnerOnly(t *testing.T) {
	dir := t.TempDir()
	sock := filepath.Join(dir, "instance.sock")
	pl, err := ListenPurge(sock, "a.example", NewCache(10, 1<<20))
	if err != nil {
		t.Fatalf("ListenPurge: %v", err)
	}
	defer pl.Close()

	fi, err := os.Stat(sock)
	if err != nil {
		t.Fatalf("stat socket: %v", err)
	}
	if mode := fi.Mode().Perm(); mode != 0600 {
		t.Fatalf("socket mode = %o, want 0600", mode)
	}
}

// Close, called immediately after ListenPurge returns (before the Serve
// goroutine it started has necessarily run at all), must still unlink the
// socket file. PurgeListener.Close used to delegate to http.Server.Close,
// which only closes listeners its own Serve call has already registered in
// its internal tracking map (net/http's trackListener, set inside Serve
// itself) — a real race, since Serve runs in a goroutine ListenPurge doesn't
// wait on. Losing that race silently no-ops Close: the goroutine's later
// Serve(ln) call then Accepts on a listener nobody ever asked to stop, and
// the socket file is never removed. A caller like UnitBackend.Start/Stop
// that opens and closes a purge listener back-to-back around a single unit
// start hits this every time, not just occasionally.
func TestListenPurgeCloseImmediatelyAfterOpenRemovesSocketFile(t *testing.T) {
	dir := shortPurgeSocketDir(t)
	sock := filepath.Join(dir, "instance.sock")
	pl, err := ListenPurge(sock, "a.example", NewCache(10, 1<<20))
	if err != nil {
		t.Fatalf("ListenPurge: %v", err)
	}

	if err := pl.Close(); err != nil {
		t.Fatalf("Close: %v", err)
	}

	if _, err := os.Stat(sock); err == nil {
		t.Fatal("expected the socket file removed immediately after Close, even called before the Serve goroutine could have run")
	}
}

// Close must tear down connections already accepted before it runs, not
// just stop new ones. Closing the raw listener alone (the fix for the
// unlink race above) only stops future Accept calls — a connection the
// server already handed off to its own handler goroutine keeps right on
// serving requests, since nothing told that goroutine to stop. A client
// whose transport pooled a keep-alive connection from a purge issued before
// Close can therefore keep issuing purges after Close returns, defeating
// the whole point of closing the listener.
func TestListenPurgeCloseRefusesRequestsOnAlreadyEstablishedConnections(t *testing.T) {
	dir := shortPurgeSocketDir(t)
	sock := filepath.Join(dir, "instance.sock")
	cache := NewCache(DefaultCacheMaxEntries, DefaultCacheMaxBodyBytes)
	cache.Put(CacheKey{Host: "a.example", Method: "GET", Path: "/event"}, "", &CachedResponse{
		StatusCode: 200, Header: http.Header{}, Body: []byte("hi"), StoredAt: time.Now(), TTL: time.Minute,
	})

	pl, err := ListenPurge(sock, "a.example", cache)
	if err != nil {
		t.Fatalf("ListenPurge: %v", err)
	}

	client := dialPurgeSocket(t, sock)
	body, _ := json.Marshal(map[string]string{"path": "/event", "query": ""})

	// Establish (and, via keep-alive, pool) a connection with one successful
	// purge before Close.
	resp, err := client.Post("http://unix/purge", "application/json", bytes.NewReader(body))
	if err != nil {
		t.Fatalf("first purge POST: %v", err)
	}
	resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("first purge status = %d, want 200", resp.StatusCode)
	}

	if err := pl.Close(); err != nil {
		t.Fatalf("Close: %v", err)
	}

	// The SAME client — free to reuse its pooled connection from the first
	// request — must not be able to complete a second purge after Close.
	resp2, err := client.Post("http://unix/purge", "application/json", bytes.NewReader(body))
	if err == nil {
		resp2.Body.Close()
		t.Fatal("expected a purge over a connection established before Close to fail once Close has run, got success")
	}
}

func TestListenPurgeRemovesStaleSocketFile(t *testing.T) {
	dir := t.TempDir()
	sock := filepath.Join(dir, "instance.sock")
	if err := os.WriteFile(sock, []byte("stale"), 0600); err != nil {
		t.Fatal(err)
	}
	pl, err := ListenPurge(sock, "a.example", NewCache(10, 1<<20))
	if err != nil {
		t.Fatalf("ListenPurge should clean up a stale socket file, got: %v", err)
	}
	pl.Close()
}

// TestListenPurgeStarPurgesTheWholeHost — {"path":"*"} is the socket twin of
// a Carlos-Purge: * header (#173): every entry of the listener's own host, and
// still never another host's.
func TestListenPurgeStarPurgesTheWholeHost(t *testing.T) {
	sock := filepath.Join(t.TempDir(), "instance.sock")
	cache := NewCache(DefaultCacheMaxEntries, DefaultCacheMaxBodyBytes)
	put := func(host, path, query string) CacheKey {
		k := CacheKey{Host: host, Method: "GET", Path: path, RawQuery: query}
		cache.Put(k, "", &CachedResponse{
			StatusCode: 200, Header: http.Header{}, Body: []byte("hi"), StoredAt: time.Now(), TTL: time.Minute,
		})
		return k
	}
	mine := []CacheKey{put("a.example", "/one", ""), put("a.example", "/two", "q=1")}
	other := put("other.example", "/one", "")

	pl, err := ListenPurge(sock, "a.example", cache)
	if err != nil {
		t.Fatalf("ListenPurge: %v", err)
	}
	defer pl.Close()

	body, _ := json.Marshal(map[string]string{"path": "*"})
	resp, err := dialPurgeSocket(t, sock).Post("http://unix/purge", "application/json", bytes.NewReader(body))
	if err != nil {
		t.Fatalf("purge POST: %v", err)
	}
	resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("purge status = %d, want 200", resp.StatusCode)
	}
	for _, k := range mine {
		if _, ok := cache.get(k); ok {
			t.Fatalf("%s%s survived a whole-host purge", k.Path, k.RawQuery)
		}
	}
	if _, ok := cache.get(other); !ok {
		t.Fatal("a whole-host purge on a.example's socket reached other.example")
	}
}

// TestListenPurgeStarWithAQueryIsRefused — "*" with a query names nothing a
// request could be stored under, so it is a malformed body, not a whole-host
// purge and not an exact purge of a path literally called "*".
func TestListenPurgeStarWithAQueryIsRefused(t *testing.T) {
	sock := filepath.Join(t.TempDir(), "instance.sock")
	cache := NewCache(DefaultCacheMaxEntries, DefaultCacheMaxBodyBytes)
	k := CacheKey{Host: "a.example", Method: "GET", Path: "/one"}
	cache.Put(k, "", &CachedResponse{
		StatusCode: 200, Header: http.Header{}, Body: []byte("hi"), StoredAt: time.Now(), TTL: time.Minute,
	})

	pl, err := ListenPurge(sock, "a.example", cache)
	if err != nil {
		t.Fatalf("ListenPurge: %v", err)
	}
	defer pl.Close()

	body, _ := json.Marshal(map[string]string{"path": "*", "query": "x=1"})
	resp, err := dialPurgeSocket(t, sock).Post("http://unix/purge", "application/json", bytes.NewReader(body))
	if err != nil {
		t.Fatalf("purge POST: %v", err)
	}
	resp.Body.Close()
	if resp.StatusCode != http.StatusBadRequest {
		t.Fatalf("purge status = %d, want 400", resp.StatusCode)
	}
	if _, ok := cache.get(k); !ok {
		t.Fatal("a refused body purged the host anyway")
	}
}

func shortPurgeSocketDir(t *testing.T) string {
	t.Helper()
	dir, err := os.MkdirTemp(os.TempDir(), "purge-")
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() {
		if err := os.RemoveAll(dir); err != nil {
			t.Error(err)
		}
	})
	return dir
}