distributor/internal/publish/execute_test.go

package publish

import (
	"context"
	"fmt"
	"io"
	"testing"
	"time"

	"gitea.maximumdirect.net/eric/distributor/internal/bundle"
	"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"
	"gitea.maximumdirect.net/eric/distributor/internal/transform"
)

func TestExecuteCleansUpAfterWriteFailure(t *testing.T) {
	sourceBackend := fake.New()
	destinationBackend := &failingBackend{Backend: fake.New(), failPath: "summary.txt"}
	sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "", testutil.BundleOptions{})
	req := Request{
		PipelineID:            "reports",
		DestinationID:         "archive",
		SourceBundle:          sourceBundle,
		SourceBackend:         sourceBackend,
		DestinationBackend:    destinationBackend,
		DestinationBundlePath: "",
		Publish:               config.PublishPolicy{Source: true},
		Transfer:              config.TransferPolicy{OnDestinationSame: config.TransferActionSkip, OnDestinationOlder: config.TransferActionReplace, OnDestinationNewer: config.TransferActionSkip, OnConflict: config.TransferActionFail},
		DistributorVersion:    "test",
	}
	plan, err := Build(context.Background(), req)
	if err != nil {
		t.Fatalf("Build() error = %v", err)
	}
	err = Execute(context.Background(), req, plan)
	if err == nil {
		t.Fatal("Execute() error = nil, want error")
	}
	found, err := destinationBackend.HasAny(context.Background(), "")
	if err != nil {
		t.Fatalf("HasAny() error = %v", err)
	}
	if found {
		t.Fatal("destination has content after failed execution")
	}
}

func TestExecuteReplaceDeletesOmittedManagedOutputs(t *testing.T) {
	sourceBackend := fake.New()
	sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "", testutil.BundleOptions{
		Files: []testutil.SourceFile{{Path: "report.md", Data: "# Report\nNew.\n"}},
	})
	destinationBackend := fake.New()
	older := sourceBundle.Manifest
	older.Created = older.Created.Add(-time.Hour)
	older.Files = append([]bundle.ManifestFile(nil), testutil.ValidManifest(testutil.BundleOptions{}).Files...)
	older.Digest = bundle.BundleDigest(older.Files)
	testutil.WriteFakeDestinationState(t, destinationBackend, "", older, testutil.DestinationStateOptions{})

	req := testRequest(sourceBackend, destinationBackend, sourceBundle, config.ReconciliationModeReplace)
	plan, err := Build(context.Background(), req)
	if err != nil {
		t.Fatalf("Build() error = %v", err)
	}
	if plan.Action != ActionReplaceOlder {
		t.Fatalf("plan action = %s, want replace_older", plan.Action)
	}
	if err := Execute(context.Background(), req, plan); err != nil {
		t.Fatalf("Execute() error = %v", err)
	}
	testutil.AssertFakeFile(t, destinationBackend, "report.md", "# Report\nNew.\n")
	testutil.AssertFakeMissing(t, destinationBackend, "summary.txt")
	destinationState := readFakeState(t, destinationBackend, "")
	if got, want := len(destinationState.Outputs), 1; got != want {
		t.Fatalf("state output count = %d, want %d", got, want)
	}
	if got, want := destinationState.Reconciliation.Mode, config.ReconciliationModeReplace; got != want {
		t.Fatalf("reconciliation mode = %q, want %q", got, want)
	}
}

func TestExecuteMergeRetainsOmittedAndOverwritesManagedOutputs(t *testing.T) {
	sourceBackend := fake.New()
	sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "", testutil.BundleOptions{
		Files: []testutil.SourceFile{{Path: "report.md", Data: "# Report\nNew.\n"}},
	})
	destinationBackend := fake.New()
	older := sourceBundle.Manifest
	older.Created = older.Created.Add(-time.Hour)
	older.Files = append([]bundle.ManifestFile(nil), testutil.ValidManifest(testutil.BundleOptions{}).Files...)
	older.Digest = bundle.BundleDigest(older.Files)
	existingState := testutil.WriteFakeDestinationState(t, destinationBackend, "", older, testutil.DestinationStateOptions{})

	req := testRequest(sourceBackend, destinationBackend, sourceBundle, config.ReconciliationModeMerge)
	plan, err := Build(context.Background(), req)
	if err != nil {
		t.Fatalf("Build() error = %v", err)
	}
	if err := Execute(context.Background(), req, plan); err != nil {
		t.Fatalf("Execute() error = %v", err)
	}
	testutil.AssertFakeFile(t, destinationBackend, "report.md", "# Report\nNew.\n")
	testutil.AssertFakeFile(t, destinationBackend, "summary.txt", "old")
	destinationState := readFakeState(t, destinationBackend, "")
	outputs := outputsByPath(destinationState.Outputs)
	if got, want := len(outputs), 2; got != want {
		t.Fatalf("state output count = %d, want %d", got, want)
	}
	if !outputs["summary.txt"].CreatedAt.Equal(existingState.Outputs[1].CreatedAt) || !outputs["summary.txt"].UpdatedAt.Equal(existingState.Outputs[1].UpdatedAt) {
		t.Fatalf("retained output timestamps = %#v, want existing %#v", outputs["summary.txt"], existingState.Outputs[1])
	}
	if !outputs["report.md"].CreatedAt.Equal(existingState.Outputs[0].CreatedAt) || !outputs["report.md"].UpdatedAt.After(existingState.Outputs[0].UpdatedAt) {
		t.Fatalf("updated output timestamps = %#v, want preserved created_at and newer updated_at", outputs["report.md"])
	}
	if got, want := destinationState.Reconciliation.Mode, config.ReconciliationModeMerge; got != want {
		t.Fatalf("reconciliation mode = %q, want %q", got, want)
	}
}

func TestExecuteReplaceConflictDeletesOmittedManagedOutputs(t *testing.T) {
	sourceBackend := fake.New()
	sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "", testutil.BundleOptions{
		Files: []testutil.SourceFile{{Path: "report.md", Data: "# Report\nNew.\n"}},
	})
	destinationBackend := fake.New()
	conflict := sourceBundle.Manifest
	conflict.ID = "other.source"
	conflict.Files = append([]bundle.ManifestFile(nil), testutil.ValidManifest(testutil.BundleOptions{}).Files...)
	conflict.Digest = bundle.BundleDigest(conflict.Files)
	testutil.WriteFakeDestinationState(t, destinationBackend, "", conflict, testutil.DestinationStateOptions{})

	req := testRequest(sourceBackend, destinationBackend, sourceBundle, config.ReconciliationModeMerge)
	req.Takeover = config.TakeoverPolicy{Mode: config.TakeoverModeNever}
	req.Transfer.OnConflict = config.TransferActionReplace
	plan, err := Build(context.Background(), req)
	if err != nil {
		t.Fatalf("Build() error = %v", err)
	}
	if plan.Action != ActionReplaceConflict {
		t.Fatalf("plan action = %s, want replace_conflict", plan.Action)
	}
	if err := Execute(context.Background(), req, plan); err != nil {
		t.Fatalf("Execute() error = %v", err)
	}
	testutil.AssertFakeFile(t, destinationBackend, "report.md", "# Report\nNew.\n")
	testutil.AssertFakeMissing(t, destinationBackend, "summary.txt")
	destinationState := readFakeState(t, destinationBackend, "")
	outputs := outputsByPath(destinationState.Outputs)
	if got, want := len(outputs), 1; got != want {
		t.Fatalf("state output count = %d, want %d", got, want)
	}
	if _, ok := outputs["summary.txt"]; ok {
		t.Fatalf("state retained summary.txt after replace_conflict: %#v", destinationState.Outputs)
	}
}

func TestExecuteReplaceNewerUsesManagedReplacement(t *testing.T) {
	sourceBackend := fake.New()
	sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "", testutil.BundleOptions{
		Files: []testutil.SourceFile{{Path: "report.md", Data: "# Report\nNew.\n"}},
	})
	destinationBackend := fake.New()
	newer := sourceBundle.Manifest
	newer.Created = newer.Created.Add(time.Hour)
	newer.Files = append([]bundle.ManifestFile(nil), testutil.ValidManifest(testutil.BundleOptions{}).Files...)
	newer.Digest = bundle.BundleDigest(newer.Files)
	testutil.WriteFakeDestinationState(t, destinationBackend, "", newer, testutil.DestinationStateOptions{})

	req := testRequest(sourceBackend, destinationBackend, sourceBundle, config.ReconciliationModeReplace)
	req.Transfer.OnDestinationNewer = config.TransferActionReplace
	plan, err := Build(context.Background(), req)
	if err != nil {
		t.Fatalf("Build() error = %v", err)
	}
	if plan.Action != ActionReplaceNewer {
		t.Fatalf("plan action = %s, want replace_newer", plan.Action)
	}
	if err := Execute(context.Background(), req, plan); err != nil {
		t.Fatalf("Execute() error = %v", err)
	}
	testutil.AssertFakeFile(t, destinationBackend, "report.md", "# Report\nNew.\n")
	testutil.AssertFakeMissing(t, destinationBackend, "summary.txt")
}

func TestExecuteMergeFailsOnUnmanagedDestinationPathCollision(t *testing.T) {
	sourceBackend := fake.New()
	sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "", testutil.BundleOptions{
		Files: []testutil.SourceFile{{Path: "report.md", Data: "# Report\nNew.\n"}},
	})
	destinationBackend := fake.New()
	older := sourceBundle.Manifest
	older.Created = older.Created.Add(-time.Hour)
	older.Files = []bundle.ManifestFile{{
		Path:   "summary.txt",
		SHA256: bundle.FileDigest([]byte("Summary\n")),
		Size:   int64(len("Summary\n")),
	}}
	older.Digest = bundle.BundleDigest(older.Files)
	testutil.WriteFakeDestinationState(t, destinationBackend, "", older, testutil.DestinationStateOptions{})
	testutil.WriteFakeFile(t, destinationBackend, "report.md", "unmanaged")

	req := testRequest(sourceBackend, destinationBackend, sourceBundle, config.ReconciliationModeMerge)
	plan, err := Build(context.Background(), req)
	if err != nil {
		t.Fatalf("Build() error = %v", err)
	}
	err = Execute(context.Background(), req, plan)
	if err == nil {
		t.Fatal("Execute() error = nil, want unmanaged path collision")
	}
	testutil.AssertFakeFile(t, destinationBackend, "report.md", "unmanaged")
	testutil.AssertFakeFile(t, destinationBackend, "summary.txt", "old")
}

func TestExecuteMergeFailureCleansUpOnlyNewOutputs(t *testing.T) {
	sourceBackend := fake.New()
	sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "", testutil.BundleOptions{
		Files: []testutil.SourceFile{
			{Path: "report.md", Data: "# Report\nNew.\n"},
			{Path: "new.md", Data: "new\n"},
			{Path: "fail.md", Data: "fail\n"},
		},
	})
	destinationBackend := &failingBackend{Backend: fake.New(), failPath: "fail.md"}
	older := sourceBundle.Manifest
	older.Created = older.Created.Add(-time.Hour)
	older.Files = []bundle.ManifestFile{sourceBundle.Manifest.Files[0]}
	older.Digest = bundle.BundleDigest(older.Files)
	testutil.WriteFakeDestinationState(t, destinationBackend.Backend, "", older, testutil.DestinationStateOptions{})

	req := testRequest(sourceBackend, destinationBackend, sourceBundle, config.ReconciliationModeMerge)
	plan, err := Build(context.Background(), req)
	if err != nil {
		t.Fatalf("Build() error = %v", err)
	}
	err = Execute(context.Background(), req, plan)
	if err == nil {
		t.Fatal("Execute() error = nil, want injected failure")
	}
	testutil.AssertFakeFile(t, destinationBackend.Backend, "report.md", "# Report\nNew.\n")
	testutil.AssertFakeMissing(t, destinationBackend.Backend, "new.md")
	testutil.AssertFakeMissing(t, destinationBackend.Backend, "fail.md")
}

func TestExecuteFixedPathSupportsReconciliationModes(t *testing.T) {
	for _, mode := range []string{config.ReconciliationModeReplace, config.ReconciliationModeMerge} {
		t.Run(mode, func(t *testing.T) {
			sourceBackend := fake.New()
			sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "new", testutil.BundleOptions{
				Files: []testutil.SourceFile{{Path: "report.md", Data: "# Report\nNew.\n"}},
			})
			destinationBackend := fake.New()
			older := sourceBundle.Manifest
			older.Created = older.Created.Add(-time.Hour)
			testutil.WriteFakeDestinationState(t, destinationBackend, "", older, testutil.DestinationStateOptions{})

			req := testRequest(sourceBackend, destinationBackend, sourceBundle, mode)
			req.DestinationBundlePath = ""
			req.PathMapping = config.PathMappingFixed
			plan, err := Build(context.Background(), req)
			if err != nil {
				t.Fatalf("Build() error = %v", err)
			}
			if plan.Action != ActionReplaceOlder {
				t.Fatalf("plan action = %s, want replace_older", plan.Action)
			}
			if err := Execute(context.Background(), req, plan); err != nil {
				t.Fatalf("Execute() error = %v", err)
			}
			testutil.AssertFakeFile(t, destinationBackend, "report.md", "# Report\nNew.\n")
			destinationState := readFakeState(t, destinationBackend, "")
			if got, want := destinationState.Reconciliation.Mode, mode; got != want {
				t.Fatalf("state reconciliation mode = %q, want %q", got, want)
			}
		})
	}
}

func TestExecuteReconciliationModesHonorPublishPolicies(t *testing.T) {
	tests := []struct {
		name        string
		publish     config.PublishPolicy
		transform   config.Transform
		transformer TransformerResolver
		wantPaths   []string
	}{
		{
			name:      "source only",
			publish:   config.PublishPolicy{Source: true},
			wantPaths: []string{"report.md"},
		},
		{
			name:      "generated only",
			publish:   config.PublishPolicy{HTML: true},
			transform: config.Transform{MarkdownToHTML: &config.MarkdownToHTML{Enabled: true, Mode: config.TransformModeSidecar}},
			transformer: testResolver{transform.MarkdownToHTML: testTransformer{outputs: []transform.Output{{
				Path:       "report.html",
				SourcePath: "report.md",
				Transform:  transform.MarkdownToHTML,
				Data:       []byte("<h1>Report</h1>\n"),
				SHA256:     bundle.FileDigest([]byte("<h1>Report</h1>\n")),
				Size:       int64(len("<h1>Report</h1>\n")),
			}}}},
			wantPaths: []string{"report.html"},
		},
		{
			name:      "source and generated",
			publish:   config.PublishPolicy{Source: true, HTML: true},
			transform: config.Transform{MarkdownToHTML: &config.MarkdownToHTML{Enabled: true, Mode: config.TransformModeSidecar}},
			transformer: testResolver{transform.MarkdownToHTML: testTransformer{outputs: []transform.Output{{
				Path:       "report.html",
				SourcePath: "report.md",
				Transform:  transform.MarkdownToHTML,
				Data:       []byte("<h1>Report</h1>\n"),
				SHA256:     bundle.FileDigest([]byte("<h1>Report</h1>\n")),
				Size:       int64(len("<h1>Report</h1>\n")),
			}}}},
			wantPaths: []string{"report.md", "report.html"},
		},
	}
	for _, mode := range []string{config.ReconciliationModeReplace, config.ReconciliationModeMerge} {
		for _, tt := range tests {
			t.Run(mode+" "+tt.name, func(t *testing.T) {
				sourceBackend := fake.New()
				sourceBundle := testutil.WriteFakeSourceBundle(t, sourceBackend, "", testutil.BundleOptions{
					Files: []testutil.SourceFile{{Path: "report.md", Data: "# Report\nNew.\n"}},
				})
				destinationBackend := fake.New()
				older := sourceBundle.Manifest
				older.Created = older.Created.Add(-time.Hour)
				testutil.WriteFakeDestinationState(t, destinationBackend, "", older, testutil.DestinationStateOptions{})

				req := testRequest(sourceBackend, destinationBackend, sourceBundle, mode)
				req.Publish = tt.publish
				req.Transform = tt.transform
				req.Transformers = tt.transformer
				plan, err := Build(context.Background(), req)
				if err != nil {
					t.Fatalf("Build() error = %v", err)
				}
				if err := Execute(context.Background(), req, plan); err != nil {
					t.Fatalf("Execute() error = %v", err)
				}
				destinationState := readFakeState(t, destinationBackend, "")
				outputs := outputsByPath(destinationState.Outputs)
				for _, path := range tt.wantPaths {
					if _, ok := outputs[path]; !ok {
						t.Fatalf("state outputs = %#v, missing %s", destinationState.Outputs, path)
					}
				}
			})
		}
	}
}

type failingBackend struct {
	*fake.Backend
	failPath string
}

func (b *failingBackend) WriteFile(ctx context.Context, path string, data []byte, opts storage.WriteOptions) (storage.Entry, error) {
	if path == b.failPath {
		return storage.Entry{}, fmt.Errorf("injected write failure")
	}
	return b.Backend.WriteFile(ctx, path, data, opts)
}

func (b *failingBackend) WriteFrom(ctx context.Context, path string, r io.Reader, opts storage.WriteOptions) (storage.Entry, error) {
	if path == b.failPath {
		return storage.Entry{}, fmt.Errorf("injected write failure")
	}
	return b.Backend.WriteFrom(ctx, path, r, opts)
}

func testRequest(sourceBackend storage.Backend, destinationBackend storage.Backend, sourceBundle bundle.Bundle, reconciliationMode string) Request {
	return Request{
		PipelineID:            "reports",
		DestinationID:         "archive",
		SourceBundle:          sourceBundle,
		SourceBackend:         sourceBackend,
		DestinationBackend:    destinationBackend,
		DestinationBundlePath: sourceBundle.RootRelativePath,
		Publish:               config.PublishPolicy{Source: true},
		Reconciliation:        config.ReconciliationPolicy{Mode: reconciliationMode},
		Transfer: config.TransferPolicy{
			OnDestinationSame:  config.TransferActionSkip,
			OnDestinationOlder: config.TransferActionReplace,
			OnDestinationNewer: config.TransferActionSkip,
			OnConflict:         config.TransferActionFail,
		},
		DistributorVersion: "test",
	}
}

func readFakeState(t *testing.T, backend storage.Backend, bundlePath string) state.DistributorState {
	t.Helper()
	statePath, err := storage.StatePath(bundlePath)
	if err != nil {
		t.Fatalf("state path: %v", err)
	}
	data, err := backend.ReadFile(context.Background(), statePath)
	if err != nil {
		t.Fatalf("read state: %v", err)
	}
	destinationState, err := state.Parse(data)
	if err != nil {
		t.Fatalf("parse state: %v", err)
	}
	return destinationState
}

func outputsByPath(outputs []state.OutputFile) map[string]state.OutputFile {
	byPath := make(map[string]state.OutputFile, len(outputs))
	for _, output := range outputs {
		byPath[output.Path] = output
	}
	return byPath
}