mirror/merlin/arthur/arthur.go

package arthur

import (
	"errors"
	"fmt"
	"io/ioutil"
	"net"
	"os"
	"path/filepath"
	"sort"
	"strings"
	"text/tabwriter"
	"time"

	"git.csclub.uwaterloo.ca/public/merlin/config"
	"git.csclub.uwaterloo.ca/public/merlin/logger"
	"git.csclub.uwaterloo.ca/public/merlin/sync"
)

// Reads and parses the message sent over the accepted connection
func GetCommand(conn net.Conn) (command, repoName string) {
	command = ""
	repoName = ""

	buf, err := ioutil.ReadAll(conn)
	if err != nil {
		logger.ErrLog(err.Error())
		return
	}

	args := strings.Split(string(buf), ":")
	if len(args) >= 1 {
		command = args[0]
	}
	if len(args) >= 2 {
		repoName = args[1]
	}
	return
}

func SendAndLog(conn net.Conn, msg string) {
	logger.OutLog(msg)
	conn.Write([]byte(msg))
}

func SendStatus(conn net.Conn) {
	// Force arthur to send back time information in America/Toronto time
	location, err := time.LoadLocation("America/Toronto")
	if err != nil {
		logger.ErrLog(err)
	}

	status := tabwriter.NewWriter(conn, 5, 5, 5, ' ', 0)
	fmt.Fprintf(status, "Repository\tLast Synced\tNext Expected Sync\tRunning\n")

	keys := make([]string, 0)
	for name, _ := range config.RepoMap {
		keys = append(keys, name)
	}
	sort.Strings(keys)

	// for other ways to format the time see: https://pkg.go.dev/time#pkg-constants
	for _, name := range keys {
		repo := config.RepoMap[name]
		lastSync := repo.State.LastAttemptStartTime
		nextSync := lastSync + int64(repo.Frequency)

		fmt.Fprintf(status, "%s\t%s\t%s\t%t\n",
			name,
			time.Unix(lastSync, 0).In(location).Format(time.RFC1123),
			time.Unix(nextSync, 0).In(location).Format(time.RFC1123),
			repo.State.IsRunning,
		)
	}

	status.Flush()
}

// Attempt to force the sync of the repo
func ForceSync(conn net.Conn, repoName string) (newSync bool) {
	newSync = false

	// TODO: send repoName and every key in RepoMap to lowercase
	if repo, isInMap := config.RepoMap[repoName]; isInMap {
		logger.OutLog("Attempting to force sync of " + repoName)
		if sync.SyncIfPossible(repo) {
			conn.Write([]byte("Forced sync for " + repoName))
			newSync = true
		} else {
			SendAndLog(conn, "Could not force sync: "+repoName+" is already syncing.")
		}
	} else {
		SendAndLog(conn, repoName+" is not tracked so cannot sync")
	}
	return
}

func StartListener(connChan chan net.Conn, stopLisChan chan struct{}) {
	sockpath := config.Conf.SockPath
	// must remove old cfg.SockPath otherwise get "bind: address already in use"
	if filepath.Ext(sockpath) != ".sock" {
		panic(fmt.Errorf("socket file must end with .sock"))
	} else if _, err := os.Stat(sockpath); err == nil {
		if err := os.Remove(sockpath); err != nil {
			panic(err)
		}
	} else if !errors.Is(err, os.ErrNotExist) {
		panic(err)
	}

	ear, err := net.Listen("unix", sockpath)
	if err != nil {
		panic(err)
	}
	logger.OutLog("Listening to unix socket at " + sockpath)

	go func() {
		for {
			// Attempting to accept on a closed net.Listener will return a non-temporary error
			conn, err := ear.Accept()
			if err != nil {
				if netErr, isTemp := err.(net.Error); isTemp && netErr.Temporary() {
					logger.ErrLog("Accepted socket error: " + err.Error())
					continue
				}
				logger.ErrLog("Unhandlable socket error: " + err.Error())
				return
			}
			connChan <- conn
		}
	}()

	// TODO: check handling of multiple SIGHUP
	<-stopLisChan
	ear.Close()
}