golib/database/rowsync/compose.go

package rowsync

import (
	"context"
	"fmt"
)

// RunStore executes the fetch → plan → apply pipeline. The planner's Mode is
// the explicit operation choice: UpdateMode is insert-update only, SyncMode
// makes the source authoritative and deletes destination rows the source
// lacks. An empty source is rejected before planning unless the destination
// is also empty, in which case the run is a no-op.
func RunStore[T any](
	ctx context.Context,
	input Fetcher[T],
	output Updater[T],
	planner Planner[T],
) (Result[T], error) {
	return runPipeline(ctx, input, output, planner, nil)
}

// RunMergedStore is RunStore with an explicit merge step for partial
// sources. The planner is forced to UpdateMode: a partial (merged) source is
// never authoritative, so destination rows the source lacks survive.
func RunMergedStore[T any](
	ctx context.Context,
	input Fetcher[T],
	output Updater[T],
	merge Merge[T],
	planner Planner[T],
) (Result[T], error) {
	if merge == nil {
		return Result[T]{}, fmt.Errorf("merge function is required")
	}
	planner.Mode = UpdateMode
	return runPipeline(ctx, input, output, planner, merge)
}

func runPipeline[T any](
	ctx context.Context,
	input Fetcher[T],
	output Updater[T],
	planner Planner[T],
	merge Merge[T],
) (Result[T], error) {
	if input == nil {
		return Result[T]{}, fmt.Errorf("input fetcher is required")
	}
	if output == nil {
		return Result[T]{}, fmt.Errorf("output updater is required")
	}
	if planner.Key == nil || planner.Equal == nil {
		return Result[T]{}, fmt.Errorf("planner key and equal functions are required")
	}
	if _, err := input.Init(ctx); err != nil {
		return Result[T]{}, fmt.Errorf("init input: %w", err)
	}
	if _, err := output.Init(ctx); err != nil {
		return Result[T]{}, fmt.Errorf("init output: %w", err)
	}
	source, err := input.Fetch(ctx)
	if err != nil {
		return Result[T]{}, fmt.Errorf("fetch input: %w", err)
	}
	previous, err := output.Fetch(ctx)
	if err != nil {
		return Result[T]{}, fmt.Errorf("fetch output: %w", err)
	}
	result := Result[T]{Source: source}
	if reporter, ok := input.(InvalidReporter); ok {
		result.Invalid = reporter.InvalidCount()
	}
	if reporter, ok := input.(InvalidRowReporter); ok {
		result.InvalidRows = reporter.InvalidRows()
	}
	current := source
	if merge != nil {
		current, err = merge(previous, source)
		if err != nil {
			return Result[T]{}, fmt.Errorf("merge: %w", err)
		}
	}
	if len(current) == 0 && len(previous) == 0 {
		// Both sides empty: nothing to plan or apply.
		result.Plan = Plan[T]{Mode: planner.Mode}
		return result, nil
	}
	plan, err := planner.Build(previous, current)
	if err != nil {
		return Result[T]{}, fmt.Errorf("build plan: %w", err)
	}
	if err := output.Update(ctx, plan); err != nil {
		return Result[T]{}, fmt.Errorf("apply plan: %w", err)
	}
	result.Plan = plan
	return result, nil
}