distributor/internal/app/upload_http_integration_test.go

package app

import (
	"archive/tar"
	"bytes"
	"compress/gzip"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"io/fs"
	"net/http"
	"net/http/httptest"
	"os"
	"path/filepath"
	"strings"
	"testing"
	"time"

	"gitea.maximumdirect.net/eric/distributor/internal/config"
	"gitea.maximumdirect.net/eric/distributor/internal/ingest"
	"gitea.maximumdirect.net/eric/distributor/internal/storage"
	"gitea.maximumdirect.net/eric/distributor/internal/testutil"
)

func TestHTTPUploadPublishesTarAndGzipFanout(t *testing.T) {
	tests := []struct {
		name        string
		compressed  bool
		contentType string
	}{
		{name: "tar", contentType: ingest.ContentTypeTar},
		{name: "gzip", compressed: true, contentType: ingest.ContentTypeGzip},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			firstDestination := t.TempDir()
			secondDestination := t.TempDir()
			cfg := httpUploadIntegrationConfig(t, []httpUploadPipelineSpec{{
				id:           "reports",
				tokenEnv:     "REPORTS_TOKEN",
				stagingPath:  filepath.Join(t.TempDir(), "reports"),
				destinations: []string{firstDestination, secondDestination},
			}}, 4, 1)
			handler, err := newUploadHTTPHandler(context.Background(), cfg, uploadHTTPTestEnvironment(map[string]string{
				"REPORTS_TOKEN": "reports-secret",
			}))
			if err != nil {
				t.Fatalf("newUploadHTTPHandler() error = %v", err)
			}
			server := httptest.NewServer(handler)
			defer server.Close()

			runID := submitHTTPUpload(t, server, "reports-secret", tt.contentType, bundleArchive(t, tt.compressed, testutil.BundleOptions{}))
			record := waitForHTTPUploadStatus(t, server, runID, UploadStatusSucceeded)

			if record.Report == nil {
				t.Fatal("completed status report = nil, want run report")
			}
			if record.Report.Summary.Status != "ok" {
				t.Fatalf("summary status = %q, want ok", record.Report.Summary.Status)
			}
			if got, want := len(record.Report.Actions), 2; got != want {
				t.Fatalf("action count = %d, want %d", got, want)
			}
			assertPublishedBundle(t, firstDestination)
			assertPublishedBundle(t, secondDestination)
		})
	}
}

func TestHTTPUploadInvalidArchiveIsRejectedWithoutRunID(t *testing.T) {
	destination := t.TempDir()
	coordinator := NewUploadCoordinator(context.Background(), httpUploadIntegrationConfig(t, []httpUploadPipelineSpec{{
		id:           "reports",
		tokenEnv:     "REPORTS_TOKEN",
		stagingPath:  filepath.Join(t.TempDir(), "reports"),
		destinations: []string{destination},
	}}, 4, 1))
	handler := uploadHTTPHandler{
		coordinator:     coordinator,
		tokens:          map[string]resolvedUploadToken{"reports-secret": uploadHTTPTestToken("reports-reporter", "reports-secret", "reports")},
		uploadPipelines: pipelineIDSet([]string{"reports"}),
	}
	server := httptest.NewServer(handler)
	defer server.Close()

	status, body := postHTTPUpload(t, server, "reports-secret", ingest.ContentTypeTar, []byte("not a tar archive"))
	if status != http.StatusBadRequest {
		t.Fatalf("POST upload status = %d, want %d; body = %s", status, http.StatusBadRequest, body)
	}
	if strings.Contains(body, "run_id") || strings.Contains(body, "reports-secret") {
		t.Fatalf("invalid archive response exposed run id or token: %s", body)
	}
	if got := coordinator.QueueDepth(); got != 0 {
		t.Fatalf("queue depth = %d, want 0", got)
	}
	assertDirectoryEmpty(t, destination)
}

func TestHTTPUploadIdempotencyReturnsOriginalRunForSameBundle(t *testing.T) {
	destination := t.TempDir()
	coordinator := NewUploadCoordinator(context.Background(), httpUploadIntegrationConfig(t, []httpUploadPipelineSpec{{
		id:           "reports",
		tokenEnv:     "REPORTS_TOKEN",
		stagingPath:  filepath.Join(t.TempDir(), "reports"),
		destinations: []string{destination},
	}}, 4, 1))
	handler := uploadHTTPHandler{
		coordinator:     coordinator,
		tokens:          map[string]resolvedUploadToken{"reports-secret": uploadHTTPTestToken("reports-reporter", "reports-secret", "reports")},
		uploadPipelines: pipelineIDSet([]string{"reports"}),
	}
	server := httptest.NewServer(handler)
	defer server.Close()

	firstRunID := submitHTTPUploadWithKey(t, server, "reports-secret", ingest.ContentTypeTar, "same-key", bundleArchive(t, false, testutil.BundleOptions{}))
	waitForHTTPUploadStatus(t, server, firstRunID, UploadStatusSucceeded)

	secondRunID := submitHTTPUploadWithKey(t, server, "reports-secret", ingest.ContentTypeGzip, "same-key", bundleArchive(t, true, testutil.BundleOptions{}))
	if secondRunID != firstRunID {
		t.Fatalf("second run id = %q, want original %q", secondRunID, firstRunID)
	}
	if got := coordinator.QueueDepth(); got != 0 {
		t.Fatalf("queue depth = %d, want no duplicate run queued", got)
	}
}

func TestHTTPUploadIdempotencyReturnsConflictForDifferentBundle(t *testing.T) {
	destination := t.TempDir()
	coordinator := NewUploadCoordinator(context.Background(), httpUploadIntegrationConfig(t, []httpUploadPipelineSpec{{
		id:           "reports",
		tokenEnv:     "REPORTS_TOKEN",
		stagingPath:  filepath.Join(t.TempDir(), "reports"),
		destinations: []string{destination},
	}}, 4, 1))
	handler := uploadHTTPHandler{
		coordinator:     coordinator,
		tokens:          map[string]resolvedUploadToken{"reports-secret": uploadHTTPTestToken("reports-reporter", "reports-secret", "reports")},
		uploadPipelines: pipelineIDSet([]string{"reports"}),
	}
	server := httptest.NewServer(handler)
	defer server.Close()

	firstRunID := submitHTTPUploadWithKey(t, server, "reports-secret", ingest.ContentTypeTar, "same-key", bundleArchive(t, false, testutil.BundleOptions{}))
	waitForHTTPUploadStatus(t, server, firstRunID, UploadStatusSucceeded)

	status, body := postHTTPUploadWithKey(t, server, "reports-secret", ingest.ContentTypeTar, "same-key", bundleArchive(t, false, testutil.BundleOptions{
		ID: "weather.daily.brentwood.2026-05-31",
	}))
	if status != http.StatusConflict {
		t.Fatalf("POST upload status = %d, want %d; body = %s", status, http.StatusConflict, body)
	}
	if strings.Contains(body, "reports-secret") {
		t.Fatalf("conflict response exposed token: %s", body)
	}
}

func TestHTTPUploadOversizedArchiveIsRejectedWithoutRunID(t *testing.T) {
	destination := t.TempDir()
	stagingPath := filepath.Join(t.TempDir(), "reports")
	cfg := httpUploadIntegrationConfig(t, []httpUploadPipelineSpec{{
		id:           "reports",
		tokenEnv:     "REPORTS_TOKEN",
		stagingPath:  stagingPath,
		destinations: []string{destination},
	}}, 4, 1)
	size := config.ByteSize(4)
	cfg.Server.HTTP.MaxUploadSize = &size
	cfg.Pipelines[0].Source.Upload.MaxUploadSize = &size
	coordinator := NewUploadCoordinator(context.Background(), cfg)
	handler := uploadHTTPHandler{
		coordinator:     coordinator,
		tokens:          map[string]resolvedUploadToken{"reports-secret": uploadHTTPTestToken("reports-reporter", "reports-secret", "reports")},
		uploadPipelines: pipelineIDSet([]string{"reports"}),
	}
	server := httptest.NewServer(handler)
	defer server.Close()

	status, body := postHTTPUpload(t, server, "reports-secret", ingest.ContentTypeTar, bundleArchive(t, false, testutil.BundleOptions{}))
	if status != http.StatusRequestEntityTooLarge {
		t.Fatalf("POST upload status = %d, want %d; body = %s", status, http.StatusRequestEntityTooLarge, body)
	}
	if strings.Contains(body, "run_id") || strings.Contains(body, "reports-secret") {
		t.Fatalf("oversized response exposed run id or token: %s", body)
	}
	if got := coordinator.QueueDepth(); got != 0 {
		t.Fatalf("queue depth = %d, want 0", got)
	}
	assertDirectoryEmpty(t, stagingPath)
	assertDirectoryEmpty(t, destination)
}

func TestHTTPUploadSamePipelineRequestsSerialize(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	release := make(chan struct{})
	started := make(chan struct{}, 1)
	coordinator := newUploadCoordinator(ctx, httpUploadIntegrationConfig(t, []httpUploadPipelineSpec{{
		id:           "reports",
		tokenEnv:     "REPORTS_TOKEN",
		stagingPath:  filepath.Join(t.TempDir(), "reports"),
		destinations: []string{t.TempDir()},
	}}, 4, 2), uploadCoordinatorHooks{
		randomSuffix: uploadTestSuffixes("00000001", "00000002"),
		stage:        successfulUploadStage,
		run: func(ctx context.Context, cfg config.Config, options RunPipelineWithLocalSourceOptions) (RunReport, error) {
			select {
			case started <- struct{}{}:
			default:
			}
			<-release
			return RunReport{}, nil
		},
	})
	handler := uploadHTTPHandler{
		coordinator:     coordinator,
		tokens:          map[string]resolvedUploadToken{"reports-secret": uploadHTTPTestToken("reports-reporter", "reports-secret", "reports")},
		uploadPipelines: pipelineIDSet([]string{"reports"}),
	}
	server := httptest.NewServer(handler)
	defer server.Close()

	firstRunID := submitHTTPUpload(t, server, "reports-secret", ingest.ContentTypeTar, []byte("first"))
	waitForRunStart(t, started)
	first := waitForHTTPUploadStatus(t, server, firstRunID, UploadStatusRunning)
	secondRunID := submitHTTPUpload(t, server, "reports-secret", ingest.ContentTypeTar, []byte("second"))
	second := waitForHTTPUploadStatus(t, server, secondRunID, UploadStatusQueued)

	if first.PipelineID != "reports" || second.PipelineID != "reports" {
		t.Fatalf("statuses = %#v %#v, want same pipeline", first, second)
	}
	if got := coordinator.RunningCount(); got != 1 {
		t.Fatalf("running count = %d, want 1", got)
	}
	close(release)
	waitForHTTPUploadStatus(t, server, firstRunID, UploadStatusSucceeded)
	waitForHTTPUploadStatus(t, server, secondRunID, UploadStatusSucceeded)
}

func TestHTTPUploadDifferentPipelinesRunConcurrently(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	release := make(chan struct{})
	started := make(chan string, 2)
	coordinator := newUploadCoordinator(ctx, httpUploadIntegrationConfig(t, []httpUploadPipelineSpec{
		{
			id:           "reports-one",
			tokenEnv:     "REPORTS_ONE_TOKEN",
			stagingPath:  filepath.Join(t.TempDir(), "reports-one"),
			destinations: []string{t.TempDir()},
		},
		{
			id:           "reports-two",
			tokenEnv:     "REPORTS_TWO_TOKEN",
			stagingPath:  filepath.Join(t.TempDir(), "reports-two"),
			destinations: []string{t.TempDir()},
		},
	}, 4, 2), uploadCoordinatorHooks{
		randomSuffix: uploadTestSuffixes("00000001", "00000002"),
		stage:        successfulUploadStage,
		run: func(ctx context.Context, cfg config.Config, options RunPipelineWithLocalSourceOptions) (RunReport, error) {
			started <- options.PipelineID
			<-release
			return RunReport{}, nil
		},
	})
	handler := uploadHTTPHandler{
		coordinator: coordinator,
		tokens: map[string]resolvedUploadToken{
			"one-secret": uploadHTTPTestToken("reports-one-reporter", "one-secret", "reports-one"),
			"two-secret": uploadHTTPTestToken("reports-two-reporter", "two-secret", "reports-two"),
		},
		uploadPipelines: pipelineIDSet([]string{"reports-one", "reports-two"}),
	}
	server := httptest.NewServer(handler)
	defer server.Close()

	firstRunID := submitHTTPUploadToPipeline(t, server, "reports-one", "one-secret", ingest.ContentTypeTar, []byte("first"))
	secondRunID := submitHTTPUploadToPipeline(t, server, "reports-two", "two-secret", ingest.ContentTypeTar, []byte("second"))
	waitForStartedPipelines(t, started, "reports-one", "reports-two")
	waitForHTTPUploadStatus(t, server, firstRunID, UploadStatusRunning)
	waitForHTTPUploadStatus(t, server, secondRunID, UploadStatusRunning)

	if got := coordinator.RunningCount(); got != 2 {
		t.Fatalf("running count = %d, want 2", got)
	}
	close(release)
	waitForHTTPUploadStatus(t, server, firstRunID, UploadStatusSucceeded)
	waitForHTTPUploadStatus(t, server, secondRunID, UploadStatusSucceeded)
}

type httpUploadPipelineSpec struct {
	id           string
	tokenEnv     string
	stagingPath  string
	destinations []string
}

func httpUploadIntegrationConfig(t *testing.T, pipelines []httpUploadPipelineSpec, queueSize, maxConcurrency int) config.Config {
	t.Helper()
	size := config.ByteSize(1024 * 1024)
	retention := config.Duration(time.Minute)
	cfg := config.Config{
		Server: config.Server{HTTP: config.HTTPServer{
			Bind:           config.DefaultHTTPBind,
			StagingRoot:    t.TempDir(),
			MaxUploadSize:  &size,
			QueueSize:      queueSize,
			MaxConcurrency: maxConcurrency,
			Retention:      &retention,
		}},
	}
	for _, spec := range pipelines {
		pipeline := config.Pipeline{
			ID: spec.id,
			Source: config.Backend{
				Backend: config.BackendHTTPUpload,
				Upload: config.HTTPUpload{
					StagingPath:   spec.stagingPath,
					MaxUploadSize: &size,
				},
			},
		}
		for index, destination := range spec.destinations {
			pipeline.Destinations = append(pipeline.Destinations, config.Destination{
				ID:      fmt.Sprintf("archive-%d", index+1),
				Backend: config.BackendLocal,
				Path:    destination,
				Publish: &config.PublishPolicy{Source: true},
			})
		}
		cfg.Pipelines = append(cfg.Pipelines, pipeline)
		cfg.UploadTokens = append(cfg.UploadTokens, config.UploadToken{
			ID:             spec.id + "-reporter",
			TokenEnv:       spec.tokenEnv,
			AllowPipelines: []string{spec.id},
		})
	}
	config.ApplyDefaults(&cfg)
	return cfg
}

func submitHTTPUpload(t *testing.T, server *httptest.Server, token, contentType string, body []byte) UploadRunID {
	t.Helper()
	return submitHTTPUploadToPipeline(t, server, "reports", token, contentType, body)
}

func submitHTTPUploadToPipeline(t *testing.T, server *httptest.Server, pipelineID, token, contentType string, body []byte) UploadRunID {
	t.Helper()
	status, responseBody := postHTTPUploadToPipeline(t, server, pipelineID, token, contentType, body)
	return decodeAcceptedHTTPUpload(t, status, responseBody)
}

func submitHTTPUploadWithKey(t *testing.T, server *httptest.Server, token, contentType, key string, body []byte) UploadRunID {
	t.Helper()
	status, responseBody := postHTTPUploadWithKey(t, server, token, contentType, key, body)
	return decodeAcceptedHTTPUpload(t, status, responseBody)
}

func decodeAcceptedHTTPUpload(t *testing.T, status int, responseBody string) UploadRunID {
	t.Helper()
	if status != http.StatusAccepted {
		t.Fatalf("POST upload status = %d, want %d; body = %s", status, http.StatusAccepted, responseBody)
	}
	var accepted uploadAcceptedResponse
	if err := json.Unmarshal([]byte(responseBody), &accepted); err != nil {
		t.Fatalf("decode accepted response: %v", err)
	}
	if accepted.RunID == "" || accepted.Status != UploadStatusAccepted {
		t.Fatalf("accepted response = %#v, want run id and accepted status", accepted)
	}
	return accepted.RunID
}

func postHTTPUpload(t *testing.T, server *httptest.Server, token, contentType string, body []byte) (int, string) {
	t.Helper()
	return postHTTPUploadToPipeline(t, server, "reports", token, contentType, body)
}

func postHTTPUploadToPipeline(t *testing.T, server *httptest.Server, pipelineID, token, contentType string, body []byte) (int, string) {
	t.Helper()
	return postHTTPUploadWithKeyToPipeline(t, server, pipelineID, token, contentType, "", body)
}

func postHTTPUploadWithKey(t *testing.T, server *httptest.Server, token, contentType, key string, body []byte) (int, string) {
	t.Helper()
	return postHTTPUploadWithKeyToPipeline(t, server, "reports", token, contentType, key, body)
}

func postHTTPUploadWithKeyToPipeline(t *testing.T, server *httptest.Server, pipelineID, token, contentType, key string, body []byte) (int, string) {
	t.Helper()
	request, err := http.NewRequest(http.MethodPost, server.URL+"/v1/pipelines/"+pipelineID+"/upload", bytes.NewReader(body))
	if err != nil {
		t.Fatalf("NewRequest() error = %v", err)
	}
	request.Header.Set("Authorization", "Bearer "+token)
	request.Header.Set("Content-Type", contentType)
	if key != "" {
		request.Header.Set("Idempotency-Key", key)
	}
	response, err := server.Client().Do(request)
	if err != nil {
		t.Fatalf("POST upload error = %v", err)
	}
	defer response.Body.Close()
	data, err := io.ReadAll(response.Body)
	if err != nil {
		t.Fatalf("read response body: %v", err)
	}
	return response.StatusCode, string(data)
}

func waitForHTTPUploadStatus(t *testing.T, server *httptest.Server, runID UploadRunID, status UploadStatus) UploadRunRecord {
	t.Helper()
	deadline := time.Now().Add(3 * time.Second)
	var latest UploadRunRecord
	var latestStatus int
	for time.Now().Before(deadline) {
		latest, latestStatus = getHTTPUploadStatus(t, server, runID)
		if latestStatus == http.StatusOK && latest.Status == status {
			return latest
		}
		time.Sleep(time.Millisecond)
	}
	t.Fatalf("timed out waiting for status %s; latest HTTP status=%d record=%#v", status, latestStatus, latest)
	return UploadRunRecord{}
}

func getHTTPUploadStatus(t *testing.T, server *httptest.Server, runID UploadRunID) (UploadRunRecord, int) {
	t.Helper()
	response, err := server.Client().Get(server.URL + "/runs/" + string(runID))
	if err != nil {
		t.Fatalf("GET /runs error = %v", err)
	}
	defer response.Body.Close()
	if response.StatusCode != http.StatusOK {
		return UploadRunRecord{}, response.StatusCode
	}
	var record UploadRunRecord
	if err := json.NewDecoder(response.Body).Decode(&record); err != nil {
		t.Fatalf("decode run status: %v", err)
	}
	return record, response.StatusCode
}

func bundleArchive(t *testing.T, compressed bool, opts testutil.BundleOptions) []byte {
	t.Helper()
	root := t.TempDir()
	testutil.WriteSourceBundle(t, root, "", opts)
	return tarDirectory(t, root, compressed)
}

func tarDirectory(t *testing.T, root string, compressed bool) []byte {
	t.Helper()
	var output bytes.Buffer
	var writer io.WriteCloser = nopWriteCloser{writer: &output}
	if compressed {
		gzipWriter := gzip.NewWriter(&output)
		writer = gzipWriter
	}
	tarWriter := tar.NewWriter(writer)
	if err := filepath.WalkDir(root, func(filePath string, entry fs.DirEntry, err error) error {
		if err != nil {
			return err
		}
		if entry.IsDir() {
			return nil
		}
		relative, err := filepath.Rel(root, filePath)
		if err != nil {
			return err
		}
		data, err := os.ReadFile(filePath)
		if err != nil {
			return err
		}
		header := &tar.Header{
			Name: filepath.ToSlash(relative),
			Mode: 0o600,
			Size: int64(len(data)),
		}
		if err := tarWriter.WriteHeader(header); err != nil {
			return err
		}
		if _, err := tarWriter.Write(data); err != nil {
			return err
		}
		return nil
	}); err != nil {
		t.Fatalf("walk bundle: %v", err)
	}
	if err := tarWriter.Close(); err != nil {
		t.Fatalf("close tar: %v", err)
	}
	if err := writer.Close(); err != nil {
		t.Fatalf("close archive: %v", err)
	}
	return output.Bytes()
}

type nopWriteCloser struct {
	writer io.Writer
}

func (writer nopWriteCloser) Write(data []byte) (int, error) {
	return writer.writer.Write(data)
}

func (writer nopWriteCloser) Close() error {
	return nil
}

func assertPublishedBundle(t *testing.T, destinationRoot string) {
	t.Helper()
	testutil.AssertFile(t, filepath.Join(destinationRoot, "report.md"), "# Report\nSunny.\n")
	testutil.AssertFile(t, filepath.Join(destinationRoot, "summary.txt"), "Summary\n")
	if _, err := os.Stat(filepath.Join(destinationRoot, storage.StateFileName)); err != nil {
		t.Fatalf("destination state stat: %v", err)
	}
}

func assertDirectoryEmpty(t *testing.T, root string) {
	t.Helper()
	entries, err := os.ReadDir(root)
	if err != nil {
		t.Fatalf("ReadDir() error = %v", err)
	}
	if len(entries) != 0 {
		t.Fatalf("directory %s has %d entries, want empty", root, len(entries))
	}
}

func waitForRunStart(t *testing.T, started <-chan struct{}) {
	t.Helper()
	select {
	case <-started:
	case <-time.After(time.Second):
		t.Fatal("timed out waiting for run start")
	}
}

func waitForStartedPipelines(t *testing.T, started <-chan string, want ...string) {
	t.Helper()
	remaining := map[string]bool{}
	for _, pipelineID := range want {
		remaining[pipelineID] = true
	}
	deadline := time.After(time.Second)
	for len(remaining) > 0 {
		select {
		case pipelineID := <-started:
			delete(remaining, pipelineID)
		case <-deadline:
			t.Fatalf("timed out waiting for pipelines to start; remaining=%v", remaining)
		}
	}
}