distributor/internal/app/reconcile_state_test.go

package app

import (
	"context"
	"strings"
	"testing"
	"time"

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

func TestReconcileStateDryRunReportsMissingManagedOutputsWithoutRewrite(t *testing.T) {
	backend := fake.New()
	cfg := reconcileStateS3Config(t)
	catalog := pruneCatalogState(time.Date(2026, 6, 8, 12, 0, 0, 0, time.UTC))
	writeFakeCatalogState(t, backend, catalog)
	if err := backend.DeleteManagedOutputs(context.Background(), "", []string{"fresh.txt"}, storage.DeleteOptions{}); err != nil {
		t.Fatalf("delete managed output: %v", err)
	}
	testutil.WriteFakeFile(t, backend, "extra.txt", "unmanaged")

	report, err := reconcileStateConfigWithBackendFactory(context.Background(), cfg, ReconcileStateOptions{
		PipelineID:    "reports",
		DestinationID: "archive",
		DryRun:        true,
	}, fakeBackendFactoryProvider(t, map[string]storage.Backend{"s3:reports": backend}))
	if err != nil {
		t.Fatalf("reconcileStateConfigWithBackendFactory() error = %v", err)
	}
	if !report.WouldChange || report.Changed {
		t.Fatalf("report changed=%t would_change=%t, want dry-run pending change", report.Changed, report.WouldChange)
	}
	if got := reportPathList(report.MissingManagedOutputs); got != "fresh.txt" {
		t.Fatalf("missing outputs = %q, want fresh.txt", got)
	}
	if got := entryPathList(report.UnmanagedEntries); got != "extra.txt" {
		t.Fatalf("unmanaged entries = %q, want extra.txt", got)
	}
	repaired := readFakeCatalogState(t, backend)
	if got := strings.Join(state.CatalogManagedOutputPaths(repaired), ","); got != "old.txt,fresh.txt,html.txt" {
		t.Fatalf("state outputs = %q, want original outputs", got)
	}
}

func TestReconcileStateApplyRemovesMissingRecordsAndPreservesUnmanagedFiles(t *testing.T) {
	backend := fake.New()
	cfg := reconcileStateS3Config(t)
	writeFakeCatalogState(t, backend, pruneCatalogState(time.Date(2026, 6, 8, 12, 0, 0, 0, time.UTC)))
	if err := backend.DeleteManagedOutputs(context.Background(), "", []string{"fresh.txt"}, storage.DeleteOptions{}); err != nil {
		t.Fatalf("delete managed output: %v", err)
	}
	testutil.WriteFakeFile(t, backend, "extra.txt", "unmanaged")

	report, err := reconcileStateConfigWithBackendFactory(context.Background(), cfg, ReconcileStateOptions{
		PipelineID:    "reports",
		DestinationID: "archive",
	}, fakeBackendFactoryProvider(t, map[string]storage.Backend{"s3:reports": backend}))
	if err != nil {
		t.Fatalf("reconcileStateConfigWithBackendFactory() error = %v", err)
	}
	if !report.Changed || report.WouldChange {
		t.Fatalf("report changed=%t would_change=%t, want applied change", report.Changed, report.WouldChange)
	}
	repaired := readFakeCatalogState(t, backend)
	if err := state.ValidateCatalog(repaired); err != nil {
		t.Fatalf("ValidateCatalog() repaired state error = %v", err)
	}
	if got := strings.Join(state.CatalogManagedOutputPaths(repaired), ","); got != "old.txt,html.txt" {
		t.Fatalf("state outputs = %q, want old.txt,html.txt", got)
	}
	if repaired.SchemaVersion != state.CatalogSchemaVersion {
		t.Fatalf("state schema_version = %d, want %d", repaired.SchemaVersion, state.CatalogSchemaVersion)
	}
	testutil.AssertFakeFile(t, backend, "extra.txt", "unmanaged")
}

func TestReconcileStateInvalidStateFailsWithoutRewrite(t *testing.T) {
	backend := fake.New()
	cfg := reconcileStateS3Config(t)
	invalid := `{"schema_version":2,"pipeline_id":"reports"}`
	testutil.WriteFakeFile(t, backend, storage.StateFileName, invalid)

	_, err := reconcileStateConfigWithBackendFactory(context.Background(), cfg, ReconcileStateOptions{
		PipelineID:    "reports",
		DestinationID: "archive",
	}, fakeBackendFactoryProvider(t, map[string]storage.Backend{"s3:reports": backend}))
	if err == nil {
		t.Fatal("reconcileStateConfigWithBackendFactory() error = nil, want invalid state error")
	}
	data, readErr := backend.ReadFile(context.Background(), storage.StateFileName)
	if readErr != nil {
		t.Fatalf("read invalid state: %v", readErr)
	}
	if string(data) != invalid {
		t.Fatalf("state data = %q, want original invalid data", data)
	}
}

func TestReconcileStateOwnerScopeRepairsCurrentOwnerOnly(t *testing.T) {
	backend := fake.New()
	cfg := reconcileStateS3Config(t)
	writeFakeCatalogState(t, backend, pruneCatalogState(time.Date(2026, 6, 8, 12, 0, 0, 0, time.UTC)))
	if err := backend.DeleteManagedOutputs(context.Background(), "", []string{"old.txt", "html.txt"}, storage.DeleteOptions{}); err != nil {
		t.Fatalf("delete managed outputs: %v", err)
	}

	report, err := reconcileStateConfigWithBackendFactory(context.Background(), cfg, ReconcileStateOptions{
		PipelineID:    "reports",
		DestinationID: "archive",
	}, fakeBackendFactoryProvider(t, map[string]storage.Backend{"s3:reports": backend}))
	if err != nil {
		t.Fatalf("reconcileStateConfigWithBackendFactory() error = %v", err)
	}
	if !report.Changed {
		t.Fatal("report changed = false, want true")
	}
	repaired := readFakeCatalogState(t, backend)
	if got := strings.Join(state.CatalogManagedOutputPaths(repaired), ","); got != "fresh.txt,html.txt" {
		t.Fatalf("catalog outputs = %q, want other owner output preserved", got)
	}
}

func TestReconcileStateAllOwnersRepairsEveryOwner(t *testing.T) {
	backend := fake.New()
	cfg := reconcileStateS3Config(t)
	writeFakeCatalogState(t, backend, pruneCatalogState(time.Date(2026, 6, 8, 12, 0, 0, 0, time.UTC)))
	if err := backend.DeleteManagedOutputs(context.Background(), "", []string{"old.txt", "html.txt"}, storage.DeleteOptions{}); err != nil {
		t.Fatalf("delete managed outputs: %v", err)
	}

	report, err := reconcileStateConfigWithBackendFactory(context.Background(), cfg, ReconcileStateOptions{
		PipelineID:    "reports",
		DestinationID: "archive",
		AllOwners:     true,
	}, fakeBackendFactoryProvider(t, map[string]storage.Backend{"s3:reports": backend}))
	if err != nil {
		t.Fatalf("reconcileStateConfigWithBackendFactory() error = %v", err)
	}
	if !report.Changed || report.CheckedCount != 3 {
		t.Fatalf("report changed=%t checked=%d, want all-owner repair", report.Changed, report.CheckedCount)
	}
	repaired := readFakeCatalogState(t, backend)
	if got := strings.Join(state.CatalogManagedOutputPaths(repaired), ","); got != "fresh.txt" {
		t.Fatalf("catalog outputs = %q, want fresh.txt", got)
	}
}

func reconcileStateS3Config(t *testing.T) config.Config {
	t.Helper()
	cfg := config.Config{Pipelines: []config.Pipeline{{
		ID:     "reports",
		Source: config.Backend{Backend: config.BackendLocal, Path: t.TempDir()},
		Destinations: []config.Destination{{
			ID:      "archive",
			Backend: config.BackendS3,
			Bucket:  "reports",
		}},
	}}}
	config.ApplyDefaults(&cfg)
	return cfg
}

func reportPathList(paths []ReconcileStatePath) string {
	values := make([]string, 0, len(paths))
	for _, path := range paths {
		values = append(values, path.Path)
	}
	return strings.Join(values, ",")
}

func entryPathList(entries []ReconcileStateEntry) string {
	values := make([]string, 0, len(entries))
	for _, entry := range entries {
		values = append(values, entry.Path)
	}
	return strings.Join(values, ",")
}