scotty/internal/cli/transfer.go

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

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 cli

import (
	"errors"
	"fmt"
	"strconv"
	"sync"
	"time"

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

func NewTransferCmd[
	E models.Backend,
	I models.ImportBackend,
	R models.ListensResult | models.LovesResult,
](
	cmd *cobra.Command,
	db *storage.Database,
	entity models.Entity,
	source string,
	target string,
) (TransferCmd[E, I, R], error) {
	c := TransferCmd[E, I, R]{
		cmd:    cmd,
		db:     db,
		entity: entity,
	}
	err := c.resolveBackends(source, target)
	if err != nil {
		return c, err
	}
	return c, nil
}

type TransferCmd[E models.Backend, I models.ImportBackend, R models.ListensResult | models.LovesResult] struct {
	cmd        *cobra.Command
	db         *storage.Database
	entity     models.Entity
	sourceName string
	targetName string
	ExpBackend E
	ImpBackend I
}

func (c *TransferCmd[E, I, R]) resolveBackends(source string, target string) error {
	sourceConfig, err := config.GetService(source)
	cobra.CheckErr(err)
	targetConfig, err := config.GetService(target)
	cobra.CheckErr(err)

	// Initialize backends
	expBackend, err := backends.ResolveBackend[E](sourceConfig)
	if err != nil {
		return err
	}
	impBackend, err := backends.ResolveBackend[I](targetConfig)
	if err != nil {
		return err
	}

	c.sourceName = sourceConfig.Name
	c.targetName = targetConfig.Name
	c.ExpBackend = expBackend
	c.ImpBackend = impBackend
	return nil
}

func (c *TransferCmd[E, I, R]) Transfer(exp backends.ExportProcessor[R], imp backends.ImportProcessor[R]) error {
	fmt.Println(i18n.Tr("Transferring %s from %s to %s...", c.entity, c.sourceName, c.targetName))

	// Authenticate backends, if needed
	config := viper.GetViper()
	_, err := backends.Authenticate(c.sourceName, c.ExpBackend, *c.db, config)
	if err != nil {
		return err
	}

	_, err = backends.Authenticate(c.targetName, c.ImpBackend, *c.db, config)
	if err != nil {
		return err
	}

	// Read timestamp
	timestamp, err := c.timestamp()
	if err != nil {
		return err
	}
	printTimestamp("From timestamp: %v (%v)", timestamp)

	// 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.Process(timestamp, exportChan, exportProgress)

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

	// Update timestamp
	err = c.updateTimestamp(result, timestamp)
	if err != nil {
		return err
	}

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

	return nil
}

func (c *TransferCmd[E, I, R]) timestamp() (time.Time, error) {
	flagValue, err := c.cmd.Flags().GetString("timestamp")
	if err != nil {
		return time.Time{}, err
	}

	// No timestamp given, read from database
	if flagValue == "" {
		timestamp, err := c.db.GetImportTimestamp(c.sourceName, c.targetName, c.entity)
		return timestamp, err
	}

	// Try using given value as a Unix timestamp
	if timestamp, err := strconv.ParseInt(flagValue, 10, 64); err == nil {
		return time.Unix(timestamp, 0), nil
	}

	// Try to parse datetime string
	for _, format := range []string{time.DateTime, time.RFC3339} {
		if t, err := time.Parse(format, flagValue); err == nil {
			return t, nil
		}
	}
	return time.Time{}, errors.New(i18n.Tr("invalid timestamp string \"%v\"", flagValue))
}

func (c *TransferCmd[E, I, R]) updateTimestamp(result models.ImportResult, oldTimestamp time.Time) error {
	if result.LastTimestamp.Unix() < oldTimestamp.Unix() {
		result.LastTimestamp = oldTimestamp
	}
	printTimestamp("Latest timestamp: %v (%v)", result.LastTimestamp)
	err := c.db.SetImportTimestamp(c.sourceName, c.targetName, c.entity, result.LastTimestamp)
	return err
}

func printTimestamp(s string, t time.Time) {
	fmt.Println(i18n.Tr(s, t, strconv.FormatInt(t.Unix(), 10)))
}