scotty/cmd/common.go

/*
Copyright © 2023 Philipp Wolfer <phw@uploadedlobster.com>

This file is part of Scotty.

Scotty is free software: you can redistribute it and/or modify it under the
terms of the GNU General Public License as published by the Free Software
Foundation, either version 3 of the License, or (at your option) any later version.

Scotty is distributed in the hope that it will be useful, but WITHOUT ANY
WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
A PARTICULAR PURPOSE. See the GNU General Public License for more details.

You should have received a copy of the GNU General Public License along with
Scotty. If not, see <https://www.gnu.org/licenses/>.
*/
package cmd

import (
	"fmt"
	"sync"
	"time"

	"github.com/spf13/cobra"
	"github.com/spf13/viper"
	"go.uploadedlobster.com/scotty/internal/backends"
	"go.uploadedlobster.com/scotty/internal/models"
	"go.uploadedlobster.com/scotty/internal/storage"
)

func getConfigFromFlag(cmd *cobra.Command, flagName string) (string, *viper.Viper) {
	configName := cmd.Flag(flagName).Value.String()
	var config *viper.Viper
	servicesConfig := viper.Sub("service")
	if servicesConfig != nil {
		config = servicesConfig.Sub(configName)
	}
	if config == nil {
		cobra.CheckErr(fmt.Sprintf("Invalid source configuration \"%s\"", configName))
	}
	return configName, config
}

func getInt64FromFlag(cmd *cobra.Command, flagName string) (result int64) {
	result, err := cmd.Flags().GetInt64(flagName)
	if err != nil {
		result = 0
	}
	return
}

type backendInfo[T models.Backend, R models.ListensResult | models.LovesResult] struct {
	configName string
	backend    T
}

type exportBackendInfo[T models.Backend, R models.ListensResult | models.LovesResult] struct {
	backendInfo[T, R]
	processor backends.ExportProcessor[R]
}

type importBackendInfo[T models.Backend, R models.ListensResult | models.LovesResult] struct {
	backendInfo[T, R]
	processor backends.ImportProcessor[R]
}

func resolveBackends[E models.Backend, I models.ImportBackend, R models.ListensResult | models.LovesResult](cmd *cobra.Command) (*exportBackendInfo[E, R], *importBackendInfo[I, R], error) {
	sourceName, sourceConfig := getConfigFromFlag(cmd, "from")
	targetName, targetConfig := getConfigFromFlag(cmd, "to")
	// Initialize backends
	exportBackend, err := backends.ResolveBackend[E](sourceConfig)
	if err != nil {
		return nil, nil, err
	}
	importBackend, err := backends.ResolveBackend[I](targetConfig)
	if err != nil {
		return nil, nil, err
	}

	exportInfo := exportBackendInfo[E, R]{
		backendInfo: backendInfo[E, R]{
			configName: sourceName,
			backend:    exportBackend,
		},
	}

	importInfo := importBackendInfo[I, R]{
		backendInfo: backendInfo[I, R]{
			configName: targetName,
			backend:    importBackend,
		},
	}

	return &exportInfo, &importInfo, nil
}

func cmdExportImport[E models.Backend, I models.ImportBackend, R models.ListensResult | models.LovesResult](cmd *cobra.Command, entity string, exp *exportBackendInfo[E, R], imp *importBackendInfo[I, R]) {
	sourceName := exp.configName
	targetName := imp.configName
	fmt.Printf("Transferring %s from %s to %s...\n", entity, sourceName, targetName)

	// Setup database
	db, err := storage.New(viper.GetString("database"))
	cobra.CheckErr(err)

	// Authenticate backends, if needed
	config := viper.GetViper()
	_, err = backends.Authenticate(sourceName, exp.backend, db, config)
	cobra.CheckErr(err)

	_, err = backends.Authenticate(targetName, imp.backend, db, config)
	cobra.CheckErr(err)

	// Read timestamp
	timestamp := time.Unix(getInt64FromFlag(cmd, "timestamp"), 0)
	if timestamp == time.Unix(0, 0) {
		timestamp, err = db.GetImportTimestamp(sourceName, targetName, entity)
		cobra.CheckErr(err)
	}
	fmt.Printf("From timestamp: %v (%v)\n", timestamp, timestamp.Unix())

	// Prepare progress bars
	exportProgress := make(chan models.Progress)
	importProgress := make(chan models.Progress)
	var wg sync.WaitGroup
	progress := progressBar(&wg, exportProgress, importProgress)

	// Export from source
	exportChan := make(chan R, 1000)
	go exp.processor.Process(timestamp, exportChan, exportProgress)

	// Import into target
	resultChan := make(chan models.ImportResult)
	go imp.processor.Process(exportChan, resultChan, importProgress)
	result := <-resultChan
	close(exportProgress)
	wg.Wait()
	progress.Wait()
	if result.Error != nil {
		fmt.Printf("Import failed, last reported timestamp was %v (%v)\n", result.LastTimestamp, result.LastTimestamp.Unix())
		cobra.CheckErr(result.Error)
	}
	fmt.Printf("Imported %v of %v %s into %v.\n",
		result.ImportCount, result.TotalCount, entity, targetName)

	// Update timestamp
	if result.LastTimestamp.Unix() < timestamp.Unix() {
		result.LastTimestamp = timestamp
	}
	fmt.Printf("Latest timestamp: %v (%v)\n", result.LastTimestamp, result.LastTimestamp.Unix())
	err = db.SetImportTimestamp(sourceName, targetName, entity, result.LastTimestamp)
	cobra.CheckErr(err)

	// Print errors
	if len(result.ImportErrors) > 0 {
		fmt.Printf("\nDuring the import the following errors occurred:\n")
		for _, err := range result.ImportErrors {
			fmt.Printf("Error: %v\n", err)
		}
	}
}