mirror

service.go (7316B)

---

 package service
 
 import (
 	"bytes"
 	"errors"
 	"fmt"
 	"os"
 	"os/exec"
 	"path"
 	"strconv"
 	"strings"
 	"sync"
 	"time"
 
 	"git.server.ky/slackcoder/mirror/internal"
 	"git.server.ky/slackcoder/mirror/internal/github"
 )
 
 // These commands are required by the service to operate.
 var requiredCommands = []string{
 	"git",
 	"rsync",
 }
 
 // Service data associated with each mirror.
 type mirrorRecord struct {
 	*Mirror
 	LastRun time.Time
 	NextRun time.Time
 }
 
 func (s *mirrorRecord) String() string {
 	return fmt.Sprintf("%s: %s -> %s at %s", s.Method, s.From, s.To, s.NextRun.Format(time.Kitchen))
 }
 
 type Service struct {
 	cfg *Config
 
 	mirrorsLock sync.Mutex
 	mirrors     []*mirrorRecord
 
 	stopCh chan struct{}
 }
 
 func NewService(arg *Config) (*Service, error) {
 	// Apply  defaults
 	var cfg Config
 
 	cfg.Append(&DefaultConfig)
 	cfg.Append(arg)
 
 	for _, cmd := range requiredCommands {
 		if err := internal.RequireCommand(cmd); err != nil {
 			return nil, err
 		}
 	}
 
 	srv := &Service{
 		cfg:     &cfg,
 		mirrors: nil,
 		stopCh:  make(chan struct{}),
 	}
 
 	for _, m := range cfg.Mirrors {
 		err := srv.AddMirror(m)
 		if err != nil {
 			return nil, err
 		}
 	}
 
 	return srv, nil
 }
 
 func (s *Service) AddMirror(arg *Mirror) error {
 	defer s.mirrorsLock.Unlock()
 	s.mirrorsLock.Lock()
 
 	for _, m := range s.mirrors {
 		if m.Equal(arg) {
 			return errors.New("must be unique")
 		}
 	}
 
 	record := &mirrorRecord{Mirror: arg}
 	scheduleNextRun(s.cfg, record)
 
 	s.mirrors = append(s.mirrors, record)
 
 	return nil
 }
 
 func (s *Service) scheduled(yield func(*mirrorRecord) bool) {
 	defer s.mirrorsLock.Unlock()
 	s.mirrorsLock.Lock()
 
 	now := time.Now()
 	for _, m := range s.mirrors {
 		if m.NextRun.IsZero() || m.NextRun.After(now) {
 			continue
 		}
 
 		if !yield(m) {
 			break
 		}
 	}
 }
 
 func runBash(runDir string, commands string) error {
 	cmd := exec.Command("bash", "-c", commands)
 	cmd.Dir = runDir
 
 	var stdErr bytes.Buffer
 	cmd.Stderr = &stdErr
 
 	err := cmd.Run()
 	if err != nil {
 		return errors.New(stdErr.String())
 	}
 
 	return nil
 }
 
 func (s *Service) Mirror(arg *Mirror) error {
 	if arg.From.Path == "" || arg.To.Path == "" || arg.Method == "" {
 		return fmt.Errorf("badly formatted mirror '%s'", arg.String())
 	}
 
 	if arg.To.Scheme != "" {
 		return fmt.Errorf("unsupported destination URL scheme '%s'", arg.To.Scheme)
 	}
 
 	cfgStagingMethod := s.cfg.StagingMethod
 	if arg.StagingMethod != "" {
 		cfgStagingMethod = arg.StagingMethod
 	}
 
 	cfgStagingPath := s.cfg.StagingPath
 	if arg.StagingPath != "" {
 		cfgStagingPath = arg.StagingPath
 	}
 
 	var downloadPath string
 	var err error
 
 	switch cfgStagingMethod {
 	case StagingMethodNone:
 		downloadPath = arg.To.Path
 	case StagingMethodTemporary:
 		tempPath, err := os.MkdirTemp("", "mirror.")
 		if err != nil {
 			return err
 		}
 		defer os.RemoveAll(tempPath)
 
 		downloadPath = tempPath
 	case StagingMethodPersistent:
 		downloadPath = path.Join(cfgStagingPath, arg.To.URL.Path)
 	default:
 		return fmt.Errorf("unknown staging method '%s'", cfgStagingMethod)
 	}
 
 	err = os.MkdirAll(downloadPath, 0755)
 	if err != nil {
 		return fmt.Errorf("creating directory '%s' for downloaded contents: %w", downloadPath, err)
 	}
 	err = os.MkdirAll(arg.To.URL.Path, 0755)
 	if err != nil {
 		return fmt.Errorf("creating directory '%s' for downloaded contents: %w", downloadPath, err)
 	}
 
 	err = Rsync(internal.MustURL(downloadPath), arg.To)
 	if err != nil {
 		return fmt.Errorf("copying current mirror data for staging to '%s': %w", downloadPath, err)
 	}
 
 	switch arg.Method {
 	case "git":
 		err = MirrorGit(internal.MustURL(downloadPath), arg.From, arg.Description)
 	case "github-assets":
 		client := github.NewClient()
 		err = client.MirrorAssets(internal.MustURL(downloadPath), arg.From)
 	case "rsync":
 		err = Rsync(internal.MustURL(downloadPath), arg.From)
 	default:
 		err = fmt.Errorf("unknown method '%s'", arg.Method)
 	}
 
 	if err != nil {
 		return fmt.Errorf("could not clone from '%s': %w", arg.From, err)
 	}
 
 	if arg.Verify != "" {
 		err = runBash(downloadPath, arg.Verify)
 		if err != nil {
 			return fmt.Errorf("verification failed: %w", err)
 		}
 	}
 
 	if downloadPath != arg.To.String() {
 		err = Rsync(arg.To, internal.MustURL(downloadPath))
 		if err != nil {
 			return fmt.Errorf("committing staged mirror data: %w", err)
 		}
 	}
 
 	return nil
 }
 
 func (s *Service) RemoveMirror(arg *Mirror) error {
 	defer s.mirrorsLock.Unlock()
 	s.mirrorsLock.Lock()
 
 	index := -1
 
 	for i, m := range s.mirrors {
 		if m.Equal(arg) {
 			index = i
 			break
 		}
 	}
 
 	if index == -1 {
 		return errors.New("not found")
 	}
 
 	s.mirrors[index] = s.mirrors[len(s.mirrors)-1]
 	s.mirrors = s.mirrors[:len(s.mirrors)-1]
 
 	return nil
 }
 
 func scheduleNextRun(cfg *Config, m *mirrorRecord) {
 	if !m.NextRun.IsZero() {
 		return
 	}
 
 	// Default to it not existing, leaving error handling for others.
 	toPathExists := false
 	_, err := os.Stat(m.To.Path)
 	if err == nil {
 		toPathExists = true
 	}
 
 	minInterval := *cfg.MinInterval
 	if m.MinInterval != nil {
 		minInterval = *m.MinInterval
 	}
 	maxInterval := *cfg.MaxInterval
 	if m.MaxInterval != nil {
 		maxInterval = *m.MaxInterval
 	}
 
 	if minInterval.Duration > 0 || maxInterval.Duration > 0 || !toPathExists {
 		lastRun := m.LastRun
 		if lastRun.IsZero() {
 			lastRun = time.Now()
 		}
 
 		m.NextRun = lastRun.Add(RandomDuration(minInterval, maxInterval).Duration)
 	}
 }
 
 // Reload reloads the service with the given configuration.
 func (s *Service) Reload(arg *Config) {
 	s.mirrorsLock.Lock()
 	defer s.mirrorsLock.Unlock()
 
 	var cfg Config
 
 	cfg.Append(&DefaultConfig)
 	cfg.Append(arg)
 
 	s.cfg = &cfg
 
 	mirrorsByDest := make(map[string]*mirrorRecord)
 	for _, m := range s.mirrors {
 		mirrorsByDest[m.To.String()] = m
 	}
 
 	mirrors := make([]*mirrorRecord, 0)
 
 	for _, m := range cfg.Mirrors {
 		record := &mirrorRecord{Mirror: m}
 
 		oldM, ok := mirrorsByDest[m.To.String()]
 		if ok {
 			delete(mirrorsByDest, m.To.String())
 
 			record.LastRun = oldM.LastRun
 			record.NextRun = oldM.NextRun
 
 			// The run times may be out of sync with the new
 			// configuration.
 			lastRun := oldM.LastRun
 			if lastRun.IsZero() {
 				lastRun = time.Now()
 			}
 
 			minInterval := *cfg.MinInterval
 			if m.MinInterval != nil {
 				minInterval = *m.MinInterval
 			}
 			maxInterval := *cfg.MaxInterval
 			if m.MaxInterval != nil {
 				maxInterval = *m.MaxInterval
 			}
 
 			waitTime := oldM.NextRun.Sub(lastRun)
 			if waitTime < minInterval.Duration || waitTime > maxInterval.Duration {
 				record.NextRun = time.Time{}
 			}
 
 		}
 
 		scheduleNextRun(s.cfg, record)
 
 		mirrors = append(
 			mirrors,
 			record,
 		)
 	}
 
 	// Swap out the old with the new.
 	s.mirrors = mirrors
 }
 
 func (s *Service) Log(str string) {
 	escaped := []rune(strconv.Quote(strings.TrimSpace(str)))
 	escaped = escaped[1 : len(escaped)-1]
 
 	fmt.Fprintf(os.Stderr, "%s: %s\n", time.Now().Format(time.RFC822Z), string(escaped))
 }
 
 func (s *Service) Run() error {
 	wakeup := time.NewTicker(time.Second)
 
 mainLoop:
 	for {
 		select {
 		case <-wakeup.C:
 		case <-s.stopCh:
 			break mainLoop
 		}
 
 		s.scheduled(func(m *mirrorRecord) bool {
 			err := s.Mirror(m.Mirror)
 			if err != nil {
 				s.Log(err.Error())
 			}
 
 			m.LastRun = time.Now()
 			m.NextRun = time.Time{}
 
 			scheduleNextRun(s.cfg, m)
 
 			return true
 		})
 	}
 
 	return nil
 }
 
 func (s *Service) Stop() {
 	select {
 	case <-s.stopCh:
 	default:
 		close(s.stopCh)
 	}
 }