package imageproxy

import (
	"os"
	"testing"
	"time"

	"sneak.berlin/go/pixa/internal/imageprocessor"
	"sneak.berlin/go/pixa/internal/imagerequest"
)

// metricValues holds the source cache, upstream fetch and transform metrics
// of a proxy and its cache. A histogram is given by its count of
// observations.
type metricValues struct {
	sourceCacheHits   uint64
	sourceCacheMisses uint64
	fetchDurations    uint64
	fetchErrors       uint64
	transformDuration uint64
	inProgress        float64
}

// readMetricValues reads the metrics of svc and its cache that
// metricValues holds from their metrics registry.
func readMetricValues(t *testing.T, svc *testProxy) metricValues {
	t.Helper()

	families, err := svc.registry.Gather()
	if err != nil {
		t.Fatalf("failed to gather the metrics: %v", err)
	}

	var values metricValues

	for _, family := range families {
		metric := family.GetMetric()[0]

		switch family.GetName() {
		case "pixa_source_cache_hits_total":
			values.sourceCacheHits = uint64(metric.GetCounter().GetValue())
		case "pixa_source_cache_misses_total":
			values.sourceCacheMisses = uint64(metric.GetCounter().GetValue())
		case "pixa_upstream_fetch_duration_seconds":
			values.fetchDurations = metric.GetHistogram().GetSampleCount()
		case "pixa_upstream_fetch_errors_total":
			values.fetchErrors = uint64(metric.GetCounter().GetValue())
		case "pixa_transform_duration_seconds":
			values.transformDuration = metric.GetHistogram().GetSampleCount()
		case "pixa_transforms_in_progress":
			values.inProgress = metric.GetGauge().GetValue()
		}
	}

	return values
}

// TestProxy_Get_CountsSourceCacheFetchesAndTransforms walks Get through a
// miss that fetches the source, a hit, a miss that uses the cached source
// and a miss whose source is not found upstream, checking the source cache,
// upstream fetch and transform metrics after each.
func TestProxy_Get_CountsSourceCacheFetchesAndTransforms(t *testing.T) {
	t.Parallel()

	svc, fixtures := setupTestProxy(t)

	// want is source cache hits, source cache misses, fetch durations
	// observed, fetch errors, transform durations observed, transforms in
	// progress.
	steps := []struct {
		name    string
		path    string
		size    int
		wantErr bool
		want    metricValues
	}{
		{"miss that fetches the source", testPathPhoto, 50, false,
			metricValues{0, 1, 1, 0, 1, 0}},
		{"hit", testPathPhoto, 50, false,
			metricValues{0, 1, 1, 0, 1, 0}},
		{"miss that uses the cached source", testPathPhoto, 25, false,
			metricValues{1, 1, 1, 0, 2, 0}},
		{"miss whose source is not found", "/images/absent.jpg", 50, true,
			metricValues{1, 2, 1, 1, 2, 0}},
	}

	for _, step := range steps {
		resp, err := svc.Get(t.Context(), &imagerequest.Request{
			Source: imagerequest.Source{
				Host: fixtures.GoodHost,
				Path: step.path,
			},
			Transform: imagerequest.Transform{
				Size:    imagerequest.Size{Width: step.size, Height: step.size},
				Format:  imagerequest.FormatJPEG,
				Quality: 85,
				FitMode: imagerequest.FitCover,
			},
		})
		if (err != nil) != step.wantErr {
			t.Fatalf("%s: Get() error = %v, want error %t", step.name, err, step.wantErr)
		}

		if err == nil {
			_ = resp.Content.Close()
		}

		got := readMetricValues(t, svc)
		t.Logf("after the %s: metrics = %+v", step.name, got)

		if got != step.want {
			t.Fatalf("after the %s: metrics = %+v, want %+v", step.name, got, step.want)
		}
	}
}

// TestProxy_Get_CountsMissingSourceFileAsSourceCacheMiss deletes the file of
// a cached source but keeps its row. A request for a second variant of that
// source must fetch the source again, and count a source cache miss, not a
// hit.
func TestProxy_Get_CountsMissingSourceFileAsSourceCacheMiss(t *testing.T) {
	t.Parallel()

	svc, fixtures := setupTestProxy(t)

	// A first request caches the photo as a source.
	resp, err := svc.Get(t.Context(), widthOnlyRequest(fixtures, 50))
	if err != nil {
		t.Fatalf("first Get() error = %v", err)
	}

	_ = resp.Content.Close()

	source := widthOnlyRequest(fixtures, 50).Source

	err = os.Remove(sourceFilePath(svc, source.Key()))
	if err != nil {
		t.Fatalf("failed to delete the cached source file: %v", err)
	}

	before := readMetricValues(t, svc)

	resp, err = svc.Get(t.Context(), widthOnlyRequest(fixtures, 25))
	if err != nil {
		t.Fatalf("Get() of a second variant error = %v", err)
	}

	_ = resp.Content.Close()

	after := readMetricValues(t, svc)
	t.Logf("before the second variant: metrics = %+v; after: %+v", before, after)

	if after.sourceCacheHits != before.sourceCacheHits ||
		after.sourceCacheMisses != before.sourceCacheMisses+1 {
		t.Errorf("source cache hits went from %d to %d and misses from %d to %d, "+
			"want no new hit and one more miss", before.sourceCacheHits,
			after.sourceCacheHits, before.sourceCacheMisses, after.sourceCacheMisses)
	}
}

// TestProxy_Get_CountsTransformWaitingForSlotAsInProgress holds the only
// processing slot and starts a request that must transform the photo. While
// the request waits for the slot, pixa_transforms_in_progress must be 1; once
// the request returns, 0.
func TestProxy_Get_CountsTransformWaitingForSlotAsInProgress(t *testing.T) {
	t.Parallel()

	svc, fixtures := setupTestProxy(t)
	svc.processor = imageprocessor.New(
		imageprocessor.Params{MaxInputBytes: 50 << 20, MaxConcurrentProcessing: 1},
	)

	release := holdProcessingSlot(t, svc.processor)

	result := startGet(t.Context(), svc, widthOnlyRequest(fixtures, 50))

	// The request fetches the photo first, then waits for the slot.
	deadline := time.Now().Add(5 * time.Second)
	for readMetricValues(t, svc).inProgress != 1 && time.Now().Before(deadline) {
		time.Sleep(10 * time.Millisecond)
	}

	waiting := readMetricValues(t, svc).inProgress
	t.Logf("while the request waits for the slot: transforms in progress = %v",
		waiting)

	select {
	case got := <-result:
		t.Fatalf("Get() returned while the only processing slot was held, "+
			"error = %v", got.err)
	default:
	}

	if waiting != 1 {
		t.Fatalf("transforms in progress = %v while the request waits for the "+
			"slot, want 1", waiting)
	}

	release()

	got := <-result
	if got.err != nil {
		t.Fatalf("Get() error = %v", got.err)
	}

	returned := readMetricValues(t, svc).inProgress
	t.Logf("once the request returned: transforms in progress = %v", returned)

	if returned != 0 {
		t.Errorf("transforms in progress = %v once the request returned, want 0",
			returned)
	}
}