package server

import (
	"context"
	stderrors "errors"
	"fmt"
	"path/filepath"
	"testing"

	"github.com/prometheus/client_golang/prometheus/testutil"
	"github.com/stretchr/testify/require"
	"go.uber.org/zap"
)

type stubLogHandler struct {
	pushFn func(ctx context.Context, fname string, targetKey string) error
}

func (h *stubLogHandler) Push(ctx context.Context, fname, targetKey string) error {
	if h.pushFn == nil {
		return nil
	}
	return h.pushFn(ctx, fname, targetKey)
}

func (h *stubLogHandler) Close() {}

func TestPendingPushFilesMetric(t *testing.T) {
	tests := []struct {
		name                   string
		mode                   Mode
		rotatedFiles           int
		limit                  int
		bucketURL              string
		pushErr                error
		wantErr                bool
		wantGauge              float64
		wantRemainingRotated   int
		verifyRemainingRotated bool
	}{
		{
			name:                   "tidb_mode_push_fail_metric_equals_backlog",
			mode:                   ModeTiDB,
			rotatedFiles:           2,
			limit:                  0,
			bucketURL:              "s3://bucket/path",
			pushErr:                stderrors.New("push failed"),
			wantErr:                true,
			wantGauge:              2,
			wantRemainingRotated:   2,
			verifyRemainingRotated: true,
		},
		{
			name:                   "agent_mode_push_fail_metric_equals_backlog",
			mode:                   ModeAgent,
			rotatedFiles:           2,
			limit:                  0,
			bucketURL:              "s3://bucket/path",
			pushErr:                stderrors.New("push failed"),
			wantErr:                true,
			wantGauge:              2,
			wantRemainingRotated:   2,
			verifyRemainingRotated: true,
		},
		{
			name:                   "tidb_mode_limit_push_updates_metric",
			mode:                   ModeTiDB,
			rotatedFiles:           4,
			limit:                  2,
			bucketURL:              "s3://bucket/path",
			pushErr:                nil,
			wantErr:                false,
			wantGauge:              2,
			wantRemainingRotated:   2,
			verifyRemainingRotated: true,
		},
		{
			name:                   "full_push_success_gauge_zero",
			mode:                   ModeTiDB,
			rotatedFiles:           3,
			limit:                  0,
			bucketURL:              "s3://bucket/path",
			pushErr:                nil,
			wantErr:                false,
			wantGauge:              0,
			wantRemainingRotated:   0,
			verifyRemainingRotated: true,
		},
		{
			name:                   "skip_push_with_empty_bucketurl_updates_gauge",
			mode:                   ModeTiDB,
			rotatedFiles:           2,
			limit:                  0,
			bucketURL:              "",
			pushErr:                nil,
			wantErr:                false,
			wantGauge:              2,
			wantRemainingRotated:   2,
			verifyRemainingRotated: true,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			// Use a non-zero sentinel to verify the metric gets updated/reset.
			pendingPushFiles.Set(123)

			dir := t.TempDir()

			files := make([]string, tt.rotatedFiles+1)
			for i := range files {
				files[i] = fmt.Sprintf("log-%d", i)
			}
			writeLog(t, filepath.Join(dir, defaultAuditLogFileName), files)

			s := &Server{
				logger: zap.NewNop(),
				Config: Config{
					LogDir:           dir,
					AuditLogFileName: defaultAuditLogFileName,
					BucketURL:        tt.bucketURL,
					PushConcurrency:  4,
					Mode:             string(tt.mode),
				},
			}

			h := &stubLogHandler{
				pushFn: func(ctx context.Context, fname string, targetKey string) error {
					return tt.pushErr
				},
			}

			_, _, err := s.pushLog(context.Background(), h, tt.limit)
			if tt.wantErr {
				require.Error(t, err)
			} else {
				require.NoError(t, err)
			}

			require.InDelta(t, tt.wantGauge, testutil.ToFloat64(pendingPushFiles), 1e-9)

			if tt.verifyRemainingRotated {
				logs, listErr := listLog(dir, defaultAuditLogFileName)
				require.NoError(t, listErr)
				require.Len(t, logs, tt.wantRemainingRotated)
			}
		})
	}
}

func TestPendingPushFilesMetricCleanLog(t *testing.T) {
	// Use a non-zero sentinel to verify the metric gets updated/reset.
	pendingPushFiles.Set(123)

	dir := t.TempDir()

	files := []string{"a", "b", "c", "d"} // 3 rotated files
	writeLog(t, filepath.Join(dir, defaultAuditLogFileName), files)

	s := &Server{
		logger: zap.NewNop(),
		Config: Config{
			LogDir:                dir,
			AuditLogFileName:      defaultAuditLogFileName,
			CleanOldestLogPercent: 0.0, // Delete all rotated logs.
		},
	}

	err := s.cleanLog()
	require.NoError(t, err)

	require.InDelta(t, float64(0), testutil.ToFloat64(pendingPushFiles), 1e-9)

	logs, listErr := listLog(dir, defaultAuditLogFileName)
	require.NoError(t, listErr)
	require.Empty(t, logs)
}