go-synapse-backupper/pkg/adapters/pipeline/pipeline.go

package pipeline

import (
	"context"
	"io"

	"git.tswf.io/infra/go-synapse-backupper/pkg/domain"
	"git.tswf.io/infra/go-synapse-backupper/pkg/domain/crypto"
	"git.tswf.io/infra/go-synapse-backupper/pkg/domain/pgdump"
)

// Runner orchestrates the dump → encrypt → sink pipeline.
type Runner interface {
	// Run executes the full backup pipeline: pg_dump → encrypt → sink.
	Run(
		ctx context.Context,
		pgDumpOpts pgdump.Options,
		recipients []crypto.RecipientPub,
		sink domain.Sink,
		rand io.Reader,
	) error
}

// Option configures a Runner.
type Option func(*runner)

// WithDumper replaces the default pg_dump dumper.
func WithDumper(dumper pgdump.Dumper) Option {
	return func(r *runner) {
		r.dumper = dumper
	}
}

// WithEncryptor replaces the default encryptor.
func WithEncryptor(encryptor crypto.Encryptor) Option {
	return func(r *runner) {
		r.encryptor = encryptor
	}
}

// runner is the private implementation of Runner.
type runner struct {
	dumper    pgdump.Dumper
	encryptor crypto.Encryptor
}

// NewRunner creates a pipeline runner with the given functional options.
func NewRunner(options ...Option) Runner {
	r := &runner{}
	for _, option := range options {
		option(r)
	}
	return r
}

// Run executes the full backup pipeline: pg_dump → encrypt → sink.
func (r *runner) Run(
	ctx context.Context,
	pgDumpOpts pgdump.Options,
	recipients []crypto.RecipientPub,
	sink domain.Sink,
	rand io.Reader,
) (retErr error) {
	tx, err := sink.Begin(pgDumpOpts.Key)
	if err != nil {
		return err
	}
	defer func() {
		if retErr != nil {
			_ = tx.Abort()
		}
	}()

	dumpCtx, dumpCancel := context.WithCancel(ctx)
	defer dumpCancel()

	pipeR, pipeW := io.Pipe()

	dumpErrCh := make(chan error, 1)
	go func() {
		dumpErrCh <- r.dumper.Dump(dumpCtx, pgDumpOpts, pipeW)
	}()

	encryptErrCh := make(chan error, 1)
	go func() {
		encryptErrCh <- r.encryptor.Encrypt(pipeR, recipients, tx, rand)
	}()

	select {
	case derr := <-dumpErrCh:
		if derr != nil {
			_ = pipeR.CloseWithError(derr)
			_ = <-encryptErrCh
			return derr
		}
		eerr := <-encryptErrCh
		if eerr != nil {
			return eerr
		}
		return tx.Commit()
	case eerr := <-encryptErrCh:
		if eerr != nil {
			dumpCancel()
			_ = pipeR.CloseWithError(eerr)
			_ = <-dumpErrCh
			return eerr
		}
		derr := <-dumpErrCh
		if derr != nil {
			return derr
		}
		return tx.Commit()
	}
}