package console

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"os"
	"sync/atomic"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/dbdoor"
	"github.com/carlosframework/platform/internal/lease"
)

type doorRangeHook struct {
	doorOpenHook
	openRange func(context.Context, string, string, string, int64, int64) (io.ReadCloser, error)
}

func (v doorRangeHook) OpenVersionRange(ctx context.Context, b, k, id string, offset, length int64) (io.ReadCloser, error) {
	return v.openRange(ctx, b, k, id, offset, length)
}

type doorCountedReader struct {
	io.ReadCloser
	ctx   context.Context
	read  *atomic.Int64
	delay time.Duration
}

func (r *doorCountedReader) Read(p []byte) (int, error) {
	if r.delay > 0 {
		timer := time.NewTimer(r.delay)
		defer timer.Stop()
		select {
		case <-r.ctx.Done():
			return 0, r.ctx.Err()
		case <-timer.C:
		}
		if len(p) > 4096 {
			p = p[:4096]
		}
	}
	n, err := r.ReadCloser.Read(p)
	r.read.Add(int64(n))
	return n, err
}

func TestDoorPlanDeepRangeReadsOnlyRequestedBytes(t *testing.T) {
	for _, ranged := range []bool{false, true} {
		t.Run(fmt.Sprint(ranged), func(t *testing.T) {
			s, g, _, model := doorPlanFixture(t, true)
			const offset = 4 << 20
			payload := append(bytes.Repeat([]byte("p"), offset), []byte("original")...)
			key := dbVersionKey(g.claim.Claim.Holder.Location, 0, 1, 1)
			id := model.Put("data", key, payload)
			model.Put("data", key, []byte("replacement"))
			var read atomic.Int64
			base := doorOpenHook{VersionedStore: model, open: func(ctx context.Context, b, k, v string) (io.ReadCloser, error) {
				rc, err := model.OpenVersion(ctx, b, k, v)
				return &doorCountedReader{ReadCloser: rc, read: &read}, err
			}}
			var store lease.VersionedStore = base
			if ranged {
				store = doorRangeHook{doorOpenHook: base, openRange: func(ctx context.Context, b, k, v string, off, length int64) (io.ReadCloser, error) {
					if b != "data" || k != key || v != id || off != offset || length != 3 {
						t.Fatalf("range arguments = %s %s %s %d %d", b, k, v, off, length)
					}
					rc, err := model.OpenVersionRange(ctx, b, k, v, off, length)
					return &doorCountedReader{ReadCloser: rc, read: &read}, err
				}}
			}
			rd := &dbVersionReader{store: store, bucket: s.cfg.HibernateBucket, handle: dbFileHandle{
				Location: g.claim.Claim.Holder.Location, Key: key, VersionID: id, Size: int64(len(payload)),
			}}
			rc, err := rd.OpenLTXFile(t.Context(), 0, 1, 1, offset, 3)
			if err != nil {
				t.Fatal(err)
			}
			body, err := io.ReadAll(rc)
			rc.Close()
			if err != nil || string(body) != "ori" {
				t.Fatalf("body = %q, %v", body, err)
			}
			want := int64(offset + 3)
			if ranged {
				want = 3
			}
			if got := read.Load(); got != want {
				t.Fatalf("storage read %d bytes; want %d", got, want)
			}
		})
	}
}

func TestDoorPlanRangeErrorsNeverFallBack(t *testing.T) {
	for _, failure := range []error{lease.ErrMirrorNotFound, lease.ErrIsDeleteMarker, context.Canceled, errors.New("offline")} {
		t.Run(failure.Error(), func(t *testing.T) {
			s, g, _, model := doorPlanFixture(t, true)
			key := dbVersionKey(g.claim.Claim.Holder.Location, 0, 1, 1)
			model.Put("data", key, []byte("replacement"))
			var fullOpens int
			store := doorRangeHook{
				doorOpenHook: doorOpenHook{VersionedStore: model, open: func(context.Context, string, string, string) (io.ReadCloser, error) {
					fullOpens++
					return io.NopCloser(bytes.NewReader([]byte("fallback"))), nil
				}},
				openRange: func(context.Context, string, string, string, int64, int64) (io.ReadCloser, error) {
					return nil, failure
				},
			}
			rd := &dbVersionReader{store: store, bucket: s.cfg.HibernateBucket, handle: dbFileHandle{
				Location: g.claim.Claim.Holder.Location, Key: key, VersionID: "selected", Size: 8,
			}}
			rc, err := rd.OpenLTXFile(t.Context(), 0, 1, 1, 2, 3)
			if rc != nil {
				rc.Close()
			}
			want := failure
			if errors.Is(failure, lease.ErrMirrorNotFound) {
				want = os.ErrNotExist
			}
			if !errors.Is(err, want) || fullOpens != 0 {
				t.Fatalf("range error = %v; want %v, full opens = %d", err, want, fullOpens)
			}
		})
	}
}

func TestDoorPlanHTTPDeepRangeKeepsWatchdogAlive(t *testing.T) {
	fx := dbFixture(t)
	model := enableDoorPlan(t, fx)
	const offset = 4 << 20
	want := bytes.Repeat([]byte("tail"), 64<<10)
	payload := append(bytes.Repeat([]byte("p"), offset), want...)
	copy(payload, doorLTXHeader(t, 1, 1))
	key := dbVersionKey("instances/"+dbTestHost, 0, 1, 1)
	id := model.Put("data-test", key, payload)
	var read, fullOpens, rangeOpens atomic.Int64
	fx.srv.dbStreamIdle = 250 * time.Millisecond
	fx.srv.cfg.DBVersions = doorRangeHook{
		doorOpenHook: doorOpenHook{VersionedStore: model, open: func(ctx context.Context, b, k, v string) (io.ReadCloser, error) {
			fullOpens.Add(1)
			rc, err := model.OpenVersion(ctx, b, k, v)
			return &doorCountedReader{ReadCloser: rc, ctx: ctx, read: &read, delay: 10 * time.Millisecond}, err
		}},
		openRange: func(ctx context.Context, b, k, v string, off, length int64) (io.ReadCloser, error) {
			rangeOpens.Add(1)
			if v != id || off != offset || length != int64(len(want)) {
				return nil, fmt.Errorf("unexpected range %s %d %d", v, off, length)
			}
			rc, err := model.OpenVersionRange(ctx, b, k, v, off, length)
			return &doorCountedReader{ReadCloser: rc, ctx: ctx, read: &read, delay: 10 * time.Millisecond}, err
		},
	}
	status, _, body := rawReq(t, http.MethodGet, fx.url(dbdoor.ListPath("bab", "titogo", dbTestHost, 0)), fx.member, "")
	var list dbdoor.ListResponse
	if err := json.Unmarshal([]byte(body), &list); err != nil || status != 200 || len(list.Files) != 1 {
		t.Fatalf("list = %d %s, %v", status, body, err)
	}
	read.Store(0)
	fullOpens.Store(0)
	model.Put("data-test", key, []byte("replacement"))
	ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second)
	defer cancel()
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, fx.url(dbdoor.GetPath("bab", "titogo", dbTestHost, 0, 1, 1)), nil)
	if err != nil {
		t.Fatal(err)
	}
	req.Header.Set("Authorization", "Bearer "+fx.member)
	req.Header.Set(dbdoor.HandleHeader, list.Files[0].Handle)
	req.Header.Set("Range", fmt.Sprintf("bytes=%d-", offset))
	start := time.Now()
	res, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	got, err := io.ReadAll(res.Body)
	res.Body.Close()
	if err != nil || res.StatusCode != 206 || !bytes.Equal(got, want) {
		t.Fatalf("range = status %d, %d bytes, %v; want 206 and %d bytes", res.StatusCode, len(got), err, len(want))
	}
	if time.Since(start) <= 2*fx.srv.dbStreamIdle {
		t.Fatal("stream did not outlive two watchdog intervals")
	}
	if read.Load() != int64(len(want)) || fullOpens.Load() != 0 || rangeOpens.Load() != 1 {
		t.Fatalf("read %d bytes, full opens %d, range opens %d", read.Load(), fullOpens.Load(), rangeOpens.Load())
	}
	if res.Header.Get("Content-Range") != fmt.Sprintf("bytes %d-%d/%d", offset, len(payload)-1, len(payload)) || res.Header.Get(dbdoor.HandleHeader) != list.Files[0].Handle {
		t.Fatalf("range headers = %v", res.Header)
	}
	fx.waitIdle(t, time.Second)
}