package aviso

import (
	"context"
	"encoding/base64"
	"encoding/json"
	"errors"
	"net/http"
	"net/http/httptest"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"
)

// pushRecorder is a stand-in push service: it records each request's
// headers and answers with whatever status the test sets.
type pushRecorder struct {
	srv    *httptest.Server
	status atomic.Int32
	hdr    chan http.Header
	retry  string
	hold   chan struct{} // when non-nil, handlers block until it closes
	inFlt  atomic.Int32
	peak   atomic.Int32
}

func newPushRecorder(t *testing.T) *pushRecorder {
	t.Helper()
	p := &pushRecorder{hdr: make(chan http.Header, 256)}
	p.status.Store(201)
	p.srv = httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		n := p.inFlt.Add(1)
		for {
			old := p.peak.Load()
			if n <= old || p.peak.CompareAndSwap(old, n) {
				break
			}
		}
		defer p.inFlt.Add(-1)
		p.hdr <- r.Header.Clone()
		if p.hold != nil {
			<-p.hold
		}
		if p.retry != "" {
			w.Header().Set("Retry-After", p.retry)
		}
		w.WriteHeader(int(p.status.Load()))
	}))
	t.Cleanup(p.srv.Close)
	return p
}

// url is the recorder's address as an endpoint Send will accept: a
// hostname, because validateEndpoint refuses a literal loopback IP
// before any HTTP happens. Only the swapped-in client (below) lets the
// name reach loopback at all.
func (p *pushRecorder) url(path string) string {
	return strings.Replace(p.srv.URL, "127.0.0.1", "localhost", 1) + path
}

// serviceAgainst is the private seam the spec names: the SSRF guard
// would refuse httptest's loopback server, so the test swaps in the
// server's own client, pinned to the certificate's name so "localhost"
// verifies. newClient's guards are tested on their own.
func serviceAgainst(t *testing.T, p *pushRecorder) *Service {
	t.Helper()
	s := newInternalService(t)
	c := p.srv.Client()
	tr := c.Transport.(*http.Transport).Clone()
	tr.TLSClientConfig.ServerName = "example.com"
	c.Transport = tr
	s.client = c
	return s
}

func jwtSub(t *testing.T, authorization string) string {
	t.Helper()
	// "vapid t=<jwt>, k=<key>"
	i := strings.Index(authorization, "t=")
	j := strings.Index(authorization, ",")
	if i < 0 || j < i {
		t.Fatalf("authorization %q", authorization)
	}
	parts := strings.Split(authorization[i+2:j], ".")
	if len(parts) != 3 {
		t.Fatalf("jwt %q", authorization)
	}
	raw, err := base64.RawURLEncoding.DecodeString(parts[1])
	if err != nil {
		t.Fatal(err)
	}
	var claims struct{ Sub string }
	if err := json.Unmarshal(raw, &claims); err != nil {
		t.Fatal(err)
	}
	return claims.Sub
}

func TestSendToSetsHeadersAndConfirms(t *testing.T) {
	p := newPushRecorder(t)
	s := serviceAgainst(t, p)
	ctx := context.Background()
	_ = s.put(ctx, "alice", sub(p.url("/one")), "")
	s.now = func() time.Time { return time.Unix(1_800_000_000, 0) }
	res, err := s.SendTo(ctx, "alice", []byte(`{"title":"hi"}`), Options{TTL: 90 * time.Second, Urgency: "high", Topic: "t1"})
	if err != nil || len(res) != 1 || res[0].Err != nil || res[0].Status != 201 {
		t.Fatalf("res=%+v err=%v", res, err)
	}
	h := <-p.hdr
	if h.Get("TTL") != "90" || h.Get("Urgency") != "high" || h.Get("Topic") != "t1" ||
		h.Get("Content-Encoding") != "aes128gcm" {
		t.Fatalf("headers: %v", h)
	}
	if sub := jwtSub(t, h.Get("Authorization")); sub != "mailto:x@y" {
		t.Fatalf("VAPID sub = %q, want mailto:x@y (webpush-go adds the prefix itself)", sub)
	}
	var confirmed int64
	_ = s.cfg.DB.QueryRow(`SELECT last_confirmed_at FROM aviso_subscriptions WHERE subject='alice'`).Scan(&confirmed)
	if confirmed != 1_800_000_000 {
		t.Fatalf("2xx did not confirm: %d", confirmed)
	}
}

func TestSendDefaultsTTLToADayAndUrgencyToNormal(t *testing.T) {
	p := newPushRecorder(t)
	s := serviceAgainst(t, p)
	ctx := context.Background()
	_ = s.put(ctx, "alice", sub(p.url("/one")), "")
	if _, err := s.SendTo(ctx, "alice", []byte("x"), Options{}); err != nil {
		t.Fatal(err)
	}
	h := <-p.hdr
	if h.Get("TTL") != "86400" || h.Get("Urgency") != "normal" || h.Get("Topic") != "" {
		t.Fatalf("headers: TTL=%q Urgency=%q Topic=%q", h.Get("TTL"), h.Get("Urgency"), h.Get("Topic"))
	}
}

func TestSendPrunesOnGoneOnlyAtSameRevision(t *testing.T) {
	p := newPushRecorder(t)
	s := serviceAgainst(t, p)
	ctx := context.Background()
	_ = s.put(ctx, "alice", sub(p.url("/one")), "")
	rows, _ := s.List(ctx, "alice")
	stale := rows[0]
	_ = s.put(ctx, "alice", sub(p.url("/one")), "") // revision 2
	p.status.Store(410)
	res, _ := s.Send(ctx, []Stored{stale}, []byte("x"), Options{})
	if res[0].Status != 410 || res[0].Err == nil {
		t.Fatalf("res=%+v", res)
	}
	if rows, _ := s.List(ctx, "alice"); len(rows) != 1 {
		t.Fatal("410 at a stale revision pruned a refreshed row")
	}
	current, _ := s.List(ctx, "alice")
	_, _ = s.Send(ctx, current, []byte("x"), Options{})
	if rows, _ := s.List(ctx, "alice"); len(rows) != 0 {
		t.Fatal("410 at the current revision did not prune")
	}
}

func TestSendReportsRetryAfter(t *testing.T) {
	p := newPushRecorder(t)
	p.retry = "120"
	p.status.Store(429)
	s := serviceAgainst(t, p)
	ctx := context.Background()
	_ = s.put(ctx, "alice", sub(p.url("/one")), "")
	res, _ := s.SendTo(ctx, "alice", []byte("x"), Options{})
	if res[0].RetryAfter != 120*time.Second || res[0].Err == nil || res[0].Status != 429 {
		t.Fatalf("res=%+v", res)
	}
	if rows, _ := s.List(ctx, "alice"); len(rows) != 1 {
		t.Fatal("429 pruned")
	}
}

func TestParseRetryAfter(t *testing.T) {
	now := time.Date(2026, 9, 7, 12, 0, 0, 0, time.UTC)
	for in, want := range map[string]time.Duration{
		"":                              0,
		"120":                           120 * time.Second,
		"-5":                            0,
		"nonsense":                      0,
		"9223372037":                    maxRetryAfter, // would overflow time.Duration
		"99999999999999999999":          maxRetryAfter,
		"Mon, 07 Sep 2026 12:05:00 GMT": 5 * time.Minute,
		"Mon, 07 Sep 2026 11:00:00 GMT": 0, // in the past
		"Wed, 07 Oct 2026 12:00:00 GMT": maxRetryAfter,
	} {
		if got := parseRetryAfter(in, now); got != want {
			t.Errorf("%q: got %v, want %v", in, got, want)
		}
	}
}

// A payload at the bound must actually encrypt and travel, not merely
// pass validation with no recipients.
func TestSendDeliversAPayloadAtTheBound(t *testing.T) {
	p := newPushRecorder(t)
	s := serviceAgainst(t, p)
	ctx := context.Background()
	_ = s.put(ctx, "alice", sub(p.url("/one")), "")
	res, err := s.SendTo(ctx, "alice", make([]byte, maxPayload), Options{})
	if err != nil || len(res) != 1 || res[0].Err != nil || res[0].Status != 201 {
		t.Fatalf("res=%+v err=%v", res, err)
	}
	<-p.hdr
}

// A failed selection is a batch error, never "zero devices".
func TestSendToReportsASelectionFailure(t *testing.T) {
	s := newInternalService(t)
	ctx := context.Background()
	_ = s.put(ctx, "alice", sub("https://push.example/one"), "")
	if _, err := s.cfg.DB.Exec(`DROP TABLE aviso_subscriptions`); err != nil {
		t.Fatal(err)
	}
	res, err := s.SendTo(ctx, "alice", []byte("x"), Options{})
	if err == nil || res != nil {
		t.Fatalf("res=%+v err=%v; want nil results and a batch error", res, err)
	}
}

func TestSendReportsServiceUnavailableAndPrunesNotFound(t *testing.T) {
	p := newPushRecorder(t)
	s := serviceAgainst(t, p)
	ctx := context.Background()
	_ = s.put(ctx, "alice", sub(p.url("/one")), "")
	p.retry = "Mon, 07 Sep 2026 12:05:00 GMT"
	p.status.Store(503)
	s.now = func() time.Time { return time.Date(2026, 9, 7, 12, 0, 0, 0, time.UTC) }
	res, _ := s.SendTo(ctx, "alice", []byte("x"), Options{})
	if res[0].Status != 503 || res[0].RetryAfter != 5*time.Minute || res[0].Err == nil {
		t.Fatalf("503: %+v", res[0])
	}
	if rows, _ := s.List(ctx, "alice"); len(rows) != 1 {
		t.Fatal("503 pruned")
	}
	p.status.Store(404)
	res, _ = s.SendTo(ctx, "alice", []byte("x"), Options{})
	if res[0].Status != 404 || res[0].Err == nil {
		t.Fatalf("404: %+v", res[0])
	}
	if rows, _ := s.List(ctx, "alice"); len(rows) != 0 {
		t.Fatal("404 did not prune")
	}
}

func TestSendSkipsRowsUnderAnotherKey(t *testing.T) {
	p := newPushRecorder(t)
	s := serviceAgainst(t, p)
	ctx := context.Background()
	_ = s.put(ctx, "alice", sub(p.url("/one")), "")
	_, _ = s.cfg.DB.Exec(`UPDATE aviso_subscriptions SET vapid_key_id = 'other'`)
	res, err := s.SendTo(ctx, "alice", []byte("x"), Options{})
	if err != nil || len(res) != 1 || !errors.Is(res[0].Err, ErrKeyMismatch) {
		t.Fatalf("res=%+v err=%v", res, err)
	}
	select {
	case <-p.hdr:
		t.Fatal("sent despite key mismatch")
	default:
	}
}

func TestSendRefusesAnEditedEndpoint(t *testing.T) {
	p := newPushRecorder(t)
	s := serviceAgainst(t, p)
	ctx := context.Background()
	_ = s.put(ctx, "alice", sub(p.url("/one")), "")
	rows, _ := s.List(ctx, "alice")
	rows[0].Endpoint = strings.Replace(rows[0].Endpoint, "https://", "http://", 1)
	res, err := s.Send(ctx, rows, []byte("x"), Options{})
	if err != nil || !errors.Is(res[0].Err, ErrBadEndpoint) {
		t.Fatalf("res=%+v err=%v", res, err)
	}
	select {
	case <-p.hdr:
		t.Fatal("sent to an http endpoint")
	default:
	}
}

func TestSendValidatesPayloadAndOptions(t *testing.T) {
	s := newInternalService(t)
	ctx := context.Background()
	if _, err := s.Send(ctx, nil, make([]byte, 3994), Options{}); !errors.Is(err, ErrPayloadTooLarge) {
		t.Errorf("oversize payload: %v", err)
	}
	if _, err := s.Send(ctx, nil, make([]byte, 3993), Options{}); err != nil {
		t.Errorf("3993 bytes refused: %v", err)
	}
	for name, o := range map[string]Options{
		"neg ttl":     {TTL: -time.Second},
		"frac ttl":    {TTL: 1500 * time.Millisecond},
		"urgency":     {Urgency: "urgent"},
		"topic chars": {Topic: "a b"},
		"topic long":  {Topic: strings.Repeat("a", 33)},
	} {
		if _, err := s.Send(ctx, nil, []byte("x"), o); !errors.Is(err, ErrBadOptions) {
			t.Errorf("%s: %v", name, err)
		}
	}
}

func TestSendHonoursCancellation(t *testing.T) {
	p := newPushRecorder(t)
	s := serviceAgainst(t, p)
	ctx, cancel := context.WithCancel(context.Background())
	cancel()
	_ = s.put(context.Background(), "alice", sub(p.url("/one")), "")
	_, err := s.SendTo(ctx, "alice", []byte("x"), Options{})
	if !errors.Is(err, context.Canceled) {
		t.Fatalf("got %v, want context.Canceled", err)
	}
}

// Cancellation while sends are in flight: rows never started carry the
// context error, started ones finish, the batch error is the context's,
// and Send returns only after every goroutine it started has stopped
// writing — which is what -race checks here.
func TestSendCancelledMidBatchWaitsForStartedGoroutines(t *testing.T) {
	p := newPushRecorder(t)
	p.hold = make(chan struct{})
	s := serviceAgainst(t, p)
	s.sem = make(chan struct{}, 2) // room for two in flight, the rest must wait
	ctx, cancel := context.WithCancel(context.Background())
	var rows []Stored
	for i := 0; i < 5; i++ {
		_ = s.put(context.Background(), "alice", sub(p.url("/"+string(rune('a'+i)))), "")
	}
	rows, _ = s.List(context.Background(), "alice")
	var wg sync.WaitGroup
	wg.Add(1)
	var res []Result
	var err error
	go func() {
		defer wg.Done()
		res, err = s.Send(ctx, rows, []byte("x"), Options{})
	}()
	<-p.hdr
	<-p.hdr // two are in flight and blocked
	cancel()
	close(p.hold)
	wg.Wait()
	if !errors.Is(err, context.Canceled) {
		t.Fatalf("batch err = %v", err)
	}
	// A row never started carries the bare context error. A row that
	// was in flight either got cut short (wrapped context error) or
	// legitimately completed before the cancellation reached the
	// transport — both are "started"; neither is the bare error.
	started, skipped := 0, 0
	for _, r := range res {
		switch {
		case r.Err == context.Canceled:
			skipped++
		case r.Err != nil || r.Status != 0:
			started++
		default:
			t.Fatalf("result neither started nor skipped: %+v", r)
		}
	}
	if started != 2 || skipped != 3 {
		t.Fatalf("started=%d skipped=%d, want 2/3", started, skipped)
	}
	if extra := len(p.hdr); extra != 0 { // the two we drained were the only requests
		t.Fatalf("recorder saw %d requests beyond the two in flight", extra)
	}
}

// Two batches on one Service share one bound: a per-call semaphore
// would let this reach 6 in flight.
func TestConcurrencyBoundIsPerService(t *testing.T) {
	p := newPushRecorder(t)
	p.hold = make(chan struct{})
	s := serviceAgainst(t, p)
	s.sem = make(chan struct{}, 3)
	ctx := context.Background()
	for i := 0; i < 6; i++ {
		_ = s.put(ctx, "alice", sub(p.url("/a"+string(rune('a'+i)))), "")
		_ = s.put(ctx, "bob", sub(p.url("/b"+string(rune('a'+i)))), "")
	}
	done := make(chan struct{}, 2)
	go func() { _, _ = s.SendTo(ctx, "alice", []byte("x"), Options{}); done <- struct{}{} }()
	go func() { _, _ = s.SendTo(ctx, "bob", []byte("x"), Options{}); done <- struct{}{} }()
	for i := 0; i < 3; i++ {
		<-p.hdr
	}
	time.Sleep(50 * time.Millisecond)
	if got := p.inFlt.Load(); got != 3 {
		t.Fatalf("in flight across two batches = %d, want 3", got)
	}
	close(p.hold)
	<-done
	<-done
	if p.peak.Load() > 3 {
		t.Fatalf("peak %d exceeded the Service bound", p.peak.Load())
	}
}

// webpush-go pads in place into the payload's spare capacity; one
// payload fanned out to many devices must not share a backing array.
// -race is what makes this test bite.
func TestSendDoesNotAliasThePayloadAcrossDevices(t *testing.T) {
	p := newPushRecorder(t)
	s := serviceAgainst(t, p)
	ctx := context.Background()
	for i := 0; i < 8; i++ {
		_ = s.put(ctx, "alice", sub(p.url("/"+string(rune('a'+i)))), "")
	}
	payload := make([]byte, 1, 4096)
	payload[0] = 'x'
	res, err := s.SendTo(ctx, "alice", payload, Options{})
	if err != nil {
		t.Fatal(err)
	}
	for _, r := range res {
		if r.Err != nil {
			t.Fatalf("%+v", r)
		}
	}
	if payload[0] != 'x' || len(payload) != 1 {
		t.Fatal("caller's payload was modified")
	}
}

// A 410 the push service already answered prunes even though the
// caller's context was cancelled by the time the answer is settled.
func TestSettlePrunesUnderACancelledContext(t *testing.T) {
	s := newInternalService(t)
	ctx, cancel := context.WithCancel(context.Background())
	_ = s.put(ctx, "alice", sub("https://push.example/one"), "")
	rows, _ := s.List(ctx, "alice")
	cancel()
	res := s.settle(ctx, rows[0], http.StatusGone, "")
	if res.Status != 410 || res.Err == nil {
		t.Fatalf("res=%+v", res)
	}
	if left, _ := s.List(context.Background(), "alice"); len(left) != 0 {
		t.Fatal("prune skipped under a cancelled context")
	}
	// And the same for confirm.
	_ = s.put(context.Background(), "alice", sub("https://push.example/two"), "")
	rows, _ = s.List(context.Background(), "alice")
	s.now = func() time.Time { return time.Unix(1_800_000_000, 0) }
	_ = s.settle(ctx, rows[0], http.StatusCreated, "")
	var confirmed int64
	_ = s.cfg.DB.QueryRow(`SELECT last_confirmed_at FROM aviso_subscriptions WHERE id=?`, rows[0].ID).Scan(&confirmed)
	if confirmed != 1_800_000_000 {
		t.Fatal("confirm skipped under a cancelled context")
	}
}

// The store update after an answer has its own bound: a transaction
// holding the single writer connection must not pin the send forever.
func TestSettleGivesUpOnABlockedWriter(t *testing.T) {
	s := newInternalService(t)
	s.dbTimeout = 100 * time.Millisecond
	ctx := context.Background()
	_ = s.put(ctx, "alice", sub("https://push.example/one"), "")
	rows, _ := s.List(ctx, "alice")
	tx, err := s.cfg.DB.BeginTx(ctx, nil)
	if err != nil {
		t.Fatal(err)
	}
	defer tx.Rollback()
	if _, err := tx.Exec(`UPDATE aviso_subscriptions SET revision = revision`); err != nil { // hold the write lock
		t.Fatal(err)
	}
	start := time.Now()
	done := make(chan Result, 1)
	go func() { done <- s.settle(ctx, rows[0], http.StatusGone, "") }()
	select {
	case res := <-done:
		if time.Since(start) > 2*time.Second {
			t.Fatalf("settle took %v with a 100ms bound", time.Since(start))
		}
		if res.Status != 410 {
			t.Fatalf("res=%+v", res)
		}
	case <-time.After(5 * time.Second):
		t.Fatal("settle blocked on the held writer")
	}
}

// A 3xx with an unparseable Location makes net/http quote the header
// verbatim in its error, and a push service could reflect the endpoint
// there. Nothing of it may survive into the Result.
func TestSendRedactsMalformedRedirects(t *testing.T) {
	p := newPushRecorder(t)
	srv := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Location", "https://push.example/secret-path/%zz")
		w.WriteHeader(http.StatusFound)
	}))
	defer srv.Close()
	p.srv = srv
	s := serviceAgainst(t, p)
	ctx := context.Background()
	_ = s.put(ctx, "alice", sub(p.url("/secret-path")), "")
	res, _ := s.SendTo(ctx, "alice", []byte("x"), Options{})
	if res[0].Err == nil || strings.Contains(res[0].Err.Error(), "secret") || strings.Contains(res[0].Err.Error(), "zz") {
		t.Fatalf("error leaks the redirect: %v", res[0].Err)
	}
	if !errors.Is(res[0].Err, ErrTransport) {
		t.Fatalf("want ErrTransport, got %v", res[0].Err)
	}
}

func TestSendBoundsConcurrency(t *testing.T) {
	p := newPushRecorder(t)
	p.hold = make(chan struct{})
	s := serviceAgainst(t, p)
	s.sem = make(chan struct{}, 3)
	ctx := context.Background()
	for i := 0; i < 8; i++ {
		_ = s.put(ctx, "alice", sub(p.url("/"+string(rune('a'+i)))), "")
	}
	done := make(chan struct{})
	go func() { _, _ = s.SendTo(ctx, "alice", []byte("x"), Options{}); close(done) }()
	for i := 0; i < 3; i++ {
		<-p.hdr
	}
	time.Sleep(50 * time.Millisecond) // give a fourth a chance to (wrongly) start
	if got := p.inFlt.Load(); got != 3 {
		t.Fatalf("in flight = %d, want 3", got)
	}
	close(p.hold)
	<-done
	if p.peak.Load() > 3 {
		t.Fatalf("peak concurrency %d exceeded bound 3", p.peak.Load())
	}
}

func TestSendRedactsEndpointFromErrors(t *testing.T) {
	s := newInternalService(t)
	ctx := context.Background()
	// Routable-looking host, refused at dial by the guard once it
	// resolves — or unreachable; either way the error must not carry
	// the URL's path, which is the secret part of an endpoint.
	_ = s.put(ctx, "alice", sub("https://127.0.0.1:9/secret-path"), "")
	rows, _ := s.List(ctx, "alice")
	rows[0].Endpoint = "https://localhost:9/secret-path" // literal IP would fail validation; a name reaches the dialer
	res, _ := s.Send(ctx, rows, []byte("x"), Options{})
	if res[0].Err == nil || strings.Contains(res[0].Err.Error(), "secret-path") {
		t.Fatalf("error carries the endpoint: %v", res[0].Err)
	}
	if !strings.Contains(res[0].Err.Error(), "aviso") {
		t.Fatalf("error not package-prefixed: %v", res[0].Err)
	}
}