package source

import (
	"context"
	"errors"
	"strings"
	"sync/atomic"
	"testing"
	"time"
)

func TestPublishLatestDropsStaleFrame(t *testing.T) {
	frames := make(chan Frame, 1)
	publishLatest(frames, Frame{Sequence: 1})
	publishLatest(frames, Frame{Sequence: 2})

	if got := <-frames; got.Sequence != 2 {
		t.Fatalf("sequence = %d, want 2", got.Sequence)
	}
}

func TestRetryDelayIsBounded(t *testing.T) {
	if got := retryDelay(0); got != time.Second {
		t.Fatalf("first retry delay = %s", got)
	}
	if got := retryDelay(3); got != 8*time.Second {
		t.Fatalf("fourth retry delay = %s", got)
	}
	if got := retryDelay(99); got != 8*time.Second {
		t.Fatalf("bounded retry delay = %s", got)
	}
}

func TestStreamRetriesAndPublishesLatestCompletedFrame(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	var decodeCalls int32
	decodedLatest := make(chan struct{})
	stream := Start(ctx, Config{SourceURL: "rtsp://operator:secret@example/101"}, Dependencies{
		Probe: func(context.Context, Config) (Metadata, error) {
			return Metadata{Width: 2, Height: 1, FPS: 30}, nil
		},
		Decode: func(ctx context.Context, _ Config, _ Metadata, publish func([]byte)) error {
			call := atomic.AddInt32(&decodeCalls, 1)
			if call == 1 {
				publish([]byte{1, 1, 1, 1, 1, 1})
				return errors.New("temporary decoder failure")
			}
			publish([]byte{2, 2, 2, 2, 2, 2})
			close(decodedLatest)
			cancel()
			return ctx.Err()
		},
		Wait: func(ctx context.Context, _ time.Duration) error { return ctx.Err() },
	})

	select {
	case <-decodedLatest:
	case <-time.After(time.Second):
		t.Fatal("decoder did not publish the second frame")
	}
	select {
	case got := <-stream.Frames():
		if got.Sequence != 2 || got.BGR[0] != 2 {
			t.Fatalf("frame = %#v", got)
		}
	case <-time.After(time.Second):
		t.Fatal("did not receive latest decoded frame")
	}
	<-stream.Done()
}

func TestStreamNeverPublishesDecoderErrorDetails(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	stream := Start(ctx, Config{SourceURL: "rtsp://operator:secret@example/101"}, Dependencies{
		Probe: func(context.Context, Config) (Metadata, error) {
			return Metadata{}, errors.New("ffprobe rtsp://operator:secret@example/101 failed")
		},
		Decode: func(context.Context, Config, Metadata, func([]byte)) error { return nil },
		Wait: func(ctx context.Context, _ time.Duration) error {
			cancel()
			return ctx.Err()
		},
	})

	for update := range stream.Statuses() {
		if strings.Contains(update.Message, "secret") || strings.Contains(update.Message, "rtsp://") {
			t.Fatalf("status leaked source details: %#v", update)
		}
	}
}