package storage

import (
	"context"
	"errors"
	"io"
	"strings"
	"sync"
	"testing"
	"time"
)

func TestFilesystemStoreCanceledPutCannotPublishLateObject(t *testing.T) {
	store, err := NewFilesystemStore(t.TempDir())
	if err != nil {
		t.Fatal(err)
	}
	reader := newBlockingFilesystemReader("partial", "late")
	ctx, cancel := context.WithCancel(t.Context())
	result := make(chan error, 1)
	go func() {
		_, putErr := store.PutObject(ctx, "attempt/file", reader, PutObjectOptions{})
		result <- putErr
	}()

	<-reader.blocked
	cancel()
	select {
	case putErr := <-result:
		if !errors.Is(putErr, context.Canceled) {
			t.Fatalf("PutObject() error = %v, want context canceled", putErr)
		}
	case <-time.After(time.Second):
		t.Fatal("PutObject did not stop after cancellation")
	}
	if _, err := store.HeadObject(t.Context(), "attempt/file"); !errors.Is(err, ErrObjectNotFound) {
		t.Fatalf("canceled object became visible: %v", err)
	}

	close(reader.release)
	<-reader.finished
	if _, err := store.HeadObject(t.Context(), "attempt/file"); !errors.Is(err, ErrObjectNotFound) {
		t.Fatalf("late reader published object after PutObject returned: %v", err)
	}
}

type blockingFilesystemReader struct {
	first, late      string
	blocked          chan struct{}
	release          chan struct{}
	finished         chan struct{}
	mu               sync.Mutex
	deliveredInitial bool
	deliveredLate    bool
}

func newBlockingFilesystemReader(first, late string) *blockingFilesystemReader {
	return &blockingFilesystemReader{
		first: first, late: late,
		blocked: make(chan struct{}), release: make(chan struct{}), finished: make(chan struct{}),
	}
}

func (r *blockingFilesystemReader) Read(p []byte) (int, error) {
	r.mu.Lock()
	if !r.deliveredInitial {
		r.deliveredInitial = true
		r.mu.Unlock()
		return strings.NewReader(r.first).Read(p)
	}
	if r.deliveredLate {
		r.mu.Unlock()
		return 0, io.EOF
	}
	r.deliveredLate = true
	r.mu.Unlock()

	close(r.blocked)
	<-r.release
	defer close(r.finished)
	return strings.NewReader(r.late).Read(p)
}