distributor/internal/app/backends.go

package app

import (
	"context"
	"fmt"
	"strconv"

	"gitea.maximumdirect.net/eric/distributor/internal/adapters/local"
	s3adapter "gitea.maximumdirect.net/eric/distributor/internal/adapters/s3"
	sshadapter "gitea.maximumdirect.net/eric/distributor/internal/adapters/ssh"
	"gitea.maximumdirect.net/eric/distributor/internal/config"
	"gitea.maximumdirect.net/eric/distributor/internal/storage"
)

const storagePathKey = "path"

const (
	sshHostKey          = "host"
	sshUserKey          = "user"
	sshPortKey          = "port"
	sshKeyFileKey       = "ssh_key_file"
	sshKnownHostsKey    = "known_hosts"
	sshHostKeyPolicyKey = "host_key_policy"
	sshReadOnlyHostsKey = "read_only_known_hosts"
	s3EndpointKey       = "endpoint"
	s3BucketKey         = "bucket"
	s3PrefixKey         = "prefix"
	s3RegionKey         = "region"
	s3ForcePathStyleKey = "force_path_style"
	s3AccessKeyIDKey    = "access_key_id"
	s3SecretAccessKey   = "secret_access_key"
)

type backendFactory struct {
	registry           *storage.Registry
	environment        config.Environment
	readOnlyKnownHosts bool
}

type backendOpenSpec struct {
	role        string
	backend     string
	path        string
	host        string
	user        string
	port        int
	ssh         config.SSH
	endpoint    string
	bucket      string
	prefix      string
	region      string
	forcePath   *bool
	credentials config.Credentials
}

func newBackendFactory() *backendFactory {
	return newBackendFactoryWithEnvironment(config.ProcessEnvironment())
}

func newBackendFactoryWithEnvironment(environment config.Environment) *backendFactory {
	registry := storage.NewRegistry()
	_ = registry.Register(config.BackendLocal, func(ctx context.Context, cfg storage.OpenConfig) (storage.Backend, error) {
		if err := ctx.Err(); err != nil {
			return nil, err
		}
		return local.New(cfg[storagePathKey])
	})
	_ = registry.Register(config.BackendSSH, func(ctx context.Context, cfg storage.OpenConfig) (storage.Backend, error) {
		port, err := strconv.Atoi(cfg[sshPortKey])
		if err != nil {
			return nil, fmt.Errorf("ssh port: %w", err)
		}
		readOnlyKnownHosts := false
		if raw := cfg[sshReadOnlyHostsKey]; raw != "" {
			readOnlyKnownHosts, err = strconv.ParseBool(raw)
			if err != nil {
				return nil, fmt.Errorf("ssh read_only_known_hosts: %w", err)
			}
		}
		return sshadapter.New(ctx, sshadapter.Options{
			Host:               cfg[sshHostKey],
			User:               cfg[sshUserKey],
			Port:               port,
			Root:               cfg[storagePathKey],
			KeyFile:            cfg[sshKeyFileKey],
			KnownHosts:         cfg[sshKnownHostsKey],
			HostKeyPolicy:      sshadapter.HostKeyPolicy(cfg[sshHostKeyPolicyKey]),
			ReadOnlyKnownHosts: readOnlyKnownHosts,
		})
	})
	_ = registry.Register(config.BackendS3, func(ctx context.Context, cfg storage.OpenConfig) (storage.Backend, error) {
		forcePathStyle, err := strconv.ParseBool(cfg[s3ForcePathStyleKey])
		if err != nil {
			return nil, fmt.Errorf("s3 force_path_style: %w", err)
		}
		return s3adapter.New(ctx, s3adapter.Options{
			Endpoint:        cfg[s3EndpointKey],
			Bucket:          cfg[s3BucketKey],
			Prefix:          cfg[s3PrefixKey],
			Region:          cfg[s3RegionKey],
			ForcePathStyle:  forcePathStyle,
			AccessKeyID:     cfg[s3AccessKeyIDKey],
			SecretAccessKey: cfg[s3SecretAccessKey],
		})
	})
	return &backendFactory{registry: registry, environment: environment}
}

func (f *backendFactory) openSource(ctx context.Context, source config.Backend) (storage.Backend, error) {
	return f.openBackend(ctx, backendOpenSpecFromSource(source))
}

func (f *backendFactory) openDestination(ctx context.Context, destination config.Destination) (storage.Backend, error) {
	return f.openBackend(ctx, backendOpenSpecFromDestination(destination))
}

func (f *backendFactory) openLocalPath(ctx context.Context, path string) (storage.Backend, error) {
	return f.registry.Open(ctx, config.BackendLocal, storage.OpenConfig{storagePathKey: path})
}

func (f *backendFactory) resolveCredentials(creds config.Credentials) (config.ResolvedCredentials, error) {
	return f.environment.ResolveCredentials(creds)
}

func (f *backendFactory) openBackend(ctx context.Context, spec backendOpenSpec) (storage.Backend, error) {
	if !backendExecutable(spec.backend) {
		if spec.role == "source" {
			return nil, fmt.Errorf("source backend %s is not implemented for execution", spec.backend)
		}
		return nil, fmt.Errorf("backend %s is not implemented for execution", spec.backend)
	}
	openConfig, err := f.openConfig(spec)
	if err != nil {
		return nil, err
	}
	return f.registry.Open(ctx, spec.backend, openConfig)
}

func backendExecutable(name string) bool {
	return name == config.BackendLocal || name == config.BackendSSH || name == config.BackendS3
}

func (f *backendFactory) openConfig(spec backendOpenSpec) (storage.OpenConfig, error) {
	cfg := storage.OpenConfig{storagePathKey: spec.path}
	switch spec.backend {
	case config.BackendSSH:
		cfg[sshHostKey] = spec.host
		cfg[sshUserKey] = spec.user
		cfg[sshPortKey] = strconv.Itoa(spec.port)
		cfg[sshKeyFileKey] = spec.ssh.KeyFile
		cfg[sshKnownHostsKey] = spec.ssh.KnownHosts
		cfg[sshHostKeyPolicyKey] = string(spec.ssh.HostKeyPolicy)
		cfg[sshReadOnlyHostsKey] = strconv.FormatBool(f.readOnlyKnownHosts)
	case config.BackendS3:
		if err := f.addS3Config(cfg, spec); err != nil {
			return nil, err
		}
	}
	return cfg, nil
}

func (f *backendFactory) addS3Config(cfg storage.OpenConfig, spec backendOpenSpec) error {
	cfg[s3EndpointKey] = spec.endpoint
	cfg[s3BucketKey] = spec.bucket
	cfg[s3PrefixKey] = spec.prefix
	cfg[s3RegionKey] = spec.region
	cfg[s3ForcePathStyleKey] = strconv.FormatBool(config.ForcePathStyle(spec.forcePath))
	if spec.credentials.AccessKeyIDEnv != "" || spec.credentials.SecretAccessKeyEnv != "" {
		resolved, err := f.resolveCredentials(spec.credentials)
		if err != nil {
			return err
		}
		cfg[s3AccessKeyIDKey] = resolved.AccessKeyID
		cfg[s3SecretAccessKey] = resolved.SecretAccessKey
	}
	return nil
}

func backendOpenSpecFromSource(source config.Backend) backendOpenSpec {
	return backendOpenSpec{
		role:        "source",
		backend:     source.Backend,
		path:        source.Path,
		host:        source.Host,
		user:        source.User,
		port:        source.Port,
		ssh:         source.SSH,
		endpoint:    source.Endpoint,
		bucket:      source.Bucket,
		prefix:      source.Prefix,
		region:      source.Region,
		forcePath:   source.ForcePath,
		credentials: source.Creds,
	}
}

func backendOpenSpecFromDestination(destination config.Destination) backendOpenSpec {
	return backendOpenSpec{
		role:        "destination",
		backend:     destination.Backend,
		path:        destination.Path,
		host:        destination.Host,
		user:        destination.User,
		port:        destination.Port,
		ssh:         destination.SSH,
		endpoint:    destination.Endpoint,
		bucket:      destination.Bucket,
		prefix:      destination.Prefix,
		region:      destination.Region,
		forcePath:   destination.ForcePath,
		credentials: destination.Creds,
	}
}