server/poll/poll.go

package poll

import (
	"time"

	"github.com/google/uuid"
	"github.com/op/go-logging"

	"git.ohea.xyz/cursorius/server/config"
	"git.ohea.xyz/cursorius/server/database"
	"git.ohea.xyz/cursorius/server/pipeline_executor"
	"github.com/go-git/go-git/v5"
	"github.com/go-git/go-git/v5/plumbing"
	"github.com/go-git/go-git/v5/storage/memory"
)

var log = logging.MustGetLogger("cursorius-server")

type branch struct {
	ref        string
	commitHash string
}

type tag struct {
	ref        string
	commitHash string
}

func pollJob(pipeline database.Pipeline, pipelineConf config.PipelineConf, db database.Database) {
	prevCommits := make(map[string]string)
	for {
		time.Sleep(time.Duration(pipeline.PollInterval) * time.Second)
		log.Infof("Polling repo %v", pipeline.Name)

		repo, err := git.Clone(memory.NewStorage(), nil, &git.CloneOptions{
			URL: pipeline.Url,
		})
		if err != nil {
			log.Errorf("Could not clone repo %v from url %v: %v", pipeline.Name, pipeline.Url, err)
			continue
		}
		refsToRunFor := []string{}

		// get branches
		branches, err := repo.Branches()
		if err != nil {
			log.Errorf("Could not enumerate branches in repo %v: %v", pipeline.Name, err)
			continue
		}

		branches.ForEach(func(branch *plumbing.Reference) error {
			log.Debugf("Processing branch %v from repo %v (id: %v)", branch.Name().String(), pipeline.Name, pipeline.Id)
			prevRef, ok := prevCommits[branch.Name().String()]
			if ok {
				if branch.Hash().String() != prevRef {
					log.Debugf("Queuing job for branch %v in repo %v (id: %v) with hash %v", branch.Name().String(), pipeline.Name, pipeline.Id, branch.Hash().String())
					prevCommits[branch.Name().String()] = branch.Hash().String()
					refsToRunFor = append(refsToRunFor, branch.Name().String())
				} else {
					log.Debugf("Branch %v in repo %v (id: %v) has hash %v, which matches the previously seen hash of %v", branch.Name().String(), pipeline.Name, pipeline.Id, branch.Hash().String(), prevRef)
				}
			} else {
				log.Debugf("Queuing job for newly discovered branch %v in repo %v (id: %v) with hash %v", branch.Name().String(), pipeline.Name, pipeline.Id, branch.Hash().String())
				prevCommits[branch.Name().String()] = branch.Hash().String()
				refsToRunFor = append(refsToRunFor, branch.Name().String())
			}
			return nil
		})

		tags, err := repo.Tags()
		if err != nil {
			log.Errorf("Could not enumerate tags in repo %v: %v", pipeline.Name, err)
			continue
		}
		tags.ForEach(func(tag *plumbing.Reference) error {
			log.Debugf("Processing tag %v from repo %v (id: %v)", tag.Name().String(), pipeline.Name, pipeline.Id)
			prevRef, ok := prevCommits[tag.Name().String()]
			if ok {
				if tag.Hash().String() != prevRef {
					log.Debugf("Queuing job for tag %v in repo %v (id: %v) with hash %v", tag.Name().String(), pipeline.Name, pipeline.Id, tag.Hash().String())
					prevCommits[tag.Name().String()] = tag.Hash().String()
					refsToRunFor = append(refsToRunFor, tag.Name().String())
				} else {
					log.Debugf("Tag %v in repo %v (id: %v) has hash %v, which matches the previously seen hash of %v", tag.Name().String(), pipeline.Name, pipeline.Id, tag.Hash().String(), prevRef)
				}
			} else {
				log.Debugf("Queuing job for newly discovered tag %v in repo %v (id: %v) with hash %v", tag.Name().String(), pipeline.Name, pipeline.Id, tag.Hash().String())
				prevCommits[tag.Name().String()] = tag.Hash().String()
				refsToRunFor = append(refsToRunFor, tag.Name().String())
			}
			return nil
		})

		for _, ref := range refsToRunFor {
			log.Debugf("Dispatching job for ref %v in repo %v (id: %v)", ref, pipeline.Name, pipeline.Id)

			run, err := db.CreateRun(pipeline.Id)
			if err != nil {
				log.Errorf("Could not create run for pipeline with id \"%v\": ", pipeline.Id, err)
				continue
			}

			pe := pipeline_executor.PipelineExecution{
				Pipeline: pipeline,
				Ref:      ref,
				Run:      run,
			}

			go pipeline_executor.ExecutePipeline(pe, db, pipelineConf)
		}
	}
}

func launchPollJobs(conf config.PipelineConf, db database.Database, pollChan chan uuid.UUID) {
	pipelines, err := db.GetPipelines()
	if err != nil {
		log.Errorf("Could not get pipelines from database: %w", err)
		return
	}

	for _, pipeline := range pipelines {
		if pipeline.PollInterval == 0 {
			continue
		} else {
			go pollJob(pipeline, conf, db)
		}
	}

	for {
		jobUUID := <-pollChan
		pipeline, err := db.GetPipelineById(jobUUID)
		if err != nil {
			log.Errorf("Could not get pipeline with id \"%v\" from database: %v", err)
			continue
		}
		// TODO: stop existing polling process for given uuid
		go pollJob(pipeline, conf, db)
	}

}

func StartPolling(conf config.PipelineConf, db database.Database) chan uuid.UUID {
	pollChan := make(chan uuid.UUID)
	go launchPollJobs(conf, db, pollChan)
	return pollChan
}