func refresh(ctx context.Context, timeout time.Duration, repo *git.Repo) error {
	ctxGitOp, cancel := context.WithTimeout(ctx, timeout)
	err := repo.Refresh(ctxGitOp)
	cancel()
	return err
}
refresh refreshes the repository, notifying the daemon we have a new
sync head.
package daemon
import (
	"bytes"
	"context"
	"crypto/sha256"
	"encoding/base64"
	"fmt"
	"time"
	"github.com/fluxcd/flux/pkg/metrics"
	"github.com/go-kit/kit/log"
	"github.com/pkg/errors"
	"github.com/fluxcd/flux/pkg/cluster"
	"github.com/fluxcd/flux/pkg/event"
	"github.com/fluxcd/flux/pkg/git"
	"github.com/fluxcd/flux/pkg/manifests"
	"github.com/fluxcd/flux/pkg/resource"
	fluxsync "github.com/fluxcd/flux/pkg/sync"
	"github.com/fluxcd/flux/pkg/update"
)
type ratchet interface {
	CurrentRevision(ctx context.Context) (string, error)
	CurrentResources() map[string]resource.Resource
	Update(ctx context.Context, oldRev, newRev string, resources map[string]resource.Resource) (bool, error)
}
type eventLogger interface {
	LogEvent(e event.Event) error
}
type changeSet struct {
	commits     []git.Commit
	oldTagRev   string
	newTagRev   string
	initialSync bool
}
func (d *Daemon) Sync(ctx context.Context, started time.Time, newRevision string, rat ratchet) error {
	ctx, cancel := context.WithTimeout(ctx, d.SyncTimeout)
	defer cancel()
	lastResources, err := d.getLastResources(ctx, rat)
	if err != nil {
		d.Logger.Log("warning", "failed to load last-synced resources. sync event may be inaccurate", "err", err)
		lastResources = map[string]resource.Resource{}
	}
	changeSet, err := d.getChangeSet(ctx, rat, newRevision)
	if err != nil {
		return err
	}
	d.Logger.Log("info", "trying to sync git changes to the cluster", "old", changeSet.oldTagRev, "new", changeSet.newTagRev)
	resourceStore, cleanup, err := d.getManifestStoreByRevision(ctx, newRevision)
	if err != nil {
		return errors.Wrap(err, "loading new resources")
	}
	defer cleanup()
	syncSetName := makeGitConfigHash(d.Repo.Origin(), d.GitConfig)
	resources, resourceErrors, err := doSync(ctx, resourceStore, d.Cluster, syncSetName, d.Logger)
	if err != nil {
		return err
	}
	updatedIDs, deletedIDs := compareResources(lastResources, resources)
	_ = deletedIDs
	notes, err := d.getNotes(ctx, d.GitTimeout)
	if err != nil {
		return err
	}
	noteEvents, includesEvents, err := d.collectNoteEvents(ctx, changeSet, notes, d.GitTimeout, started, d.Logger)
	if err != nil {
		return err
	}
	if err := logCommitEvent(d, changeSet, updatedIDs, started, includesEvents, resourceErrors, d.Logger); err != nil {
		return err
	}
	for _, event := range noteEvents {
		if err = d.LogEvent(event); err != nil {
			d.Logger.Log("err", err)
			return err
		}
	}
	if ok, err := rat.Update(ctx, changeSet.oldTagRev, changeSet.newTagRev, resources); err != nil {
		return err
	} else if !ok {
		return nil
	}
	err = refresh(ctx, d.GitTimeout, d.Repo)
	return err
}
func (d *Daemon) getLastResources(ctx context.Context, rat ratchet) (map[string]resource.Resource, error) {
	lastResources := rat.CurrentResources()
	if lastResources != nil {
		return lastResources, nil
	}
	currentRevision, err := rat.CurrentRevision(ctx)
	if err != nil {
		return nil, err
	}
	if currentRevision == "" {
		return make(map[string]resource.Resource), nil
	}
	lastResourcestore, cleanup, err := d.getManifestStoreByRevision(ctx, currentRevision)
	if err != nil {
		return nil, errors.Wrap(err, "reading the repository checkout")
	}
	defer cleanup()
	lastResources, err = lastResourcestore.GetAllResourcesByID(ctx)
	if err != nil {
		return nil, errors.Wrap(err, "loading resources from repo")
	}
	return lastResources, nil
}
func (d *Daemon) getManifestStoreByRevision(ctx context.Context, revision string) (store manifests.Store, cleanupClone func(), err error) {
	clone, cleanupClone, err := d.cloneRepo(ctx, revision)
	if err != nil {
		return nil, nil, errors.Wrap(err, "cloning repo")
	}
	store, err = d.getManifestStore(clone)
	return store, cleanupClone, err
}
func (d *Daemon) cloneRepo(ctx context.Context, revision string) (clone *git.Export, cleanup func(), err error) {
	ctxGitOp, cancel := context.WithTimeout(ctx, d.GitTimeout)
	defer cancel()
	clone, err = d.Repo.Export(ctxGitOp, revision)
	if err != nil {
		return nil, nil, err
	}
	if d.GitSecretEnabled {
		ctxGitOp, cancel := context.WithTimeout(ctx, d.GitTimeout)
		defer cancel()
		if err := clone.SecretUnseal(ctxGitOp); err != nil {
			return nil, nil, err
		}
	}
	cleanup = func() {
		if err := clone.Clean(); err != nil {
			d.Logger.Log("error", fmt.Sprintf("cannot clean clone: %s", err))
		}
	}
	return clone, cleanup, nil
}
func (d *Daemon) getChangeSet(ctxGitOp context.Context, state ratchet, headRev string) (changeSet, error) {
	var c changeSet
	var err error
	currentRev, err := state.CurrentRevision(ctxGitOp)
	if err != nil {
		return c, err
	}
	c.oldTagRev = currentRev
	c.newTagRev = headRev
	paths := d.GitConfig.Paths
	if d.ManifestGenerationEnabled {
		paths = []string{}
	}
	ctxGitOp, cancel := context.WithTimeout(ctxGitOp, d.GitTimeout)
	if c.oldTagRev != "" {
		c.commits, err = d.Repo.CommitsBetween(ctxGitOp, c.oldTagRev, c.newTagRev, false, paths...)
	} else {
		c.initialSync = true
		c.commits, err = d.Repo.CommitsBefore(ctxGitOp, c.newTagRev, false, paths...)
	}
	cancel()
	return c, err
}
func doSync(ctx context.Context, manifestsStore manifests.Store, clus cluster.Cluster, syncSetName string,
	logger log.Logger) (map[string]resource.Resource, []event.ResourceError, error) {
	resources, err := manifestsStore.GetAllResourcesByID(ctx)
	if err != nil {
		return nil, nil, errors.Wrap(err, "loading resources from repo")
	}
	var resourceErrors []event.ResourceError
	if err := fluxsync.Sync(syncSetName, resources, clus); err != nil {
		switch syncerr := err.(type) {
		case cluster.SyncError:
			logger.Log("err", err)
			updateSyncManifestsMetric(len(resources)-len(syncerr), len(syncerr))
			for _, e := range syncerr {
				resourceErrors = append(resourceErrors, event.ResourceError{
					ID:    e.ResourceID,
					Path:  e.Source,
					Error: e.Error.Error(),
				})
			}
		default:
			return nil, nil, err
		}
	} else {
		updateSyncManifestsMetric(len(resources), 0)
	}
	return resources, resourceErrors, nil
}
func updateSyncManifestsMetric(success, failure int) {
	syncManifestsMetric.With(metrics.LabelSuccess, "true").Set(float64(success))
	syncManifestsMetric.With(metrics.LabelSuccess, "false").Set(float64(failure))
}
func compareResources(old, new map[string]resource.Resource) (updated, deleted resource.IDSet) {
	updated, deleted = resource.IDSet{}, resource.IDSet{}
	toIDs := func(r resource.Resource) []resource.ID { return []resource.ID{r.ResourceID()} }
	for newID, newResource := range new {
		if oldResource, ok := old[newID]; ok {
			if !bytes.Equal(oldResource.Bytes(), newResource.Bytes()) {
				updated.Add(toIDs(newResource))
			}
		} else {
			updated.Add(toIDs(newResource))
		}
	}
	for oldID, oldResource := range old {
		if _, ok := new[oldID]; !ok {
			deleted.Add(toIDs(oldResource))
		}
	}
	return updated, deleted
}
func (d *Daemon) getNotes(ctx context.Context, timeout time.Duration) (map[string]struct{}, error) {
	ctxGitOp, cancel := context.WithTimeout(ctx, timeout)
	notes, err := d.Repo.NoteRevList(ctxGitOp, d.GitConfig.NotesRef)
	cancel()
	if err != nil {
		return nil, errors.Wrap(err, "loading notes from repo")
	}
	return notes, nil
}
func (d *Daemon) collectNoteEvents(ctx context.Context, c changeSet, notes map[string]struct{}, timeout time.Duration,
	started time.Time, logger log.Logger) ([]event.Event, map[string]bool, error) {
	if len(c.commits) == 0 {
		return nil, nil, nil
	}
	var noteEvents []event.Event
	var eventTypes = make(map[string]bool)
	for i := len(c.commits) - 1; i >= 0; i-- {
		if _, ok := notes[c.commits[i].Revision]; !ok {
			eventTypes[event.NoneOfTheAbove] = true
			continue
		}
		var n note
		ctxGitOp, cancel := context.WithTimeout(ctx, timeout)
		ok, err := d.Repo.GetNote(ctxGitOp, c.commits[i].Revision, d.GitConfig.NotesRef, &n)
		cancel()
		if err != nil {
			return nil, nil, errors.Wrap(err, "loading notes from repo")
		}
		if !ok {
			eventTypes[event.NoneOfTheAbove] = true
			continue
		}
		if c.initialSync {
			logger.Log("warning", "no notes expected on initial sync; this repo may be in use by another fluxd")
			return noteEvents, eventTypes, nil
		}
		switch n.Spec.Type {
		case update.Containers:
			spec := n.Spec.Spec.(update.ReleaseContainersSpec)
			noteEvents = append(noteEvents, event.Event{
				ServiceIDs: n.Result.AffectedResources(),
				Type:       event.EventRelease,
				StartedAt:  started,
				EndedAt:    time.Now().UTC(),
				LogLevel:   event.LogLevelInfo,
				Metadata: &event.ReleaseEventMetadata{
					ReleaseEventCommon: event.ReleaseEventCommon{
						Revision: c.commits[i].Revision,
						Result:   n.Result,
						Error:    n.Result.Error(),
					},
					Spec: event.ReleaseSpec{
						Type:                  event.ReleaseContainersSpecType,
						ReleaseContainersSpec: &spec,
					},
					Cause: n.Spec.Cause,
				},
			})
			eventTypes[event.EventRelease] = true
		case update.Images:
			spec := n.Spec.Spec.(update.ReleaseImageSpec)
			noteEvents = append(noteEvents, event.Event{
				ServiceIDs: n.Result.AffectedResources(),
				Type:       event.EventRelease,
				StartedAt:  started,
				EndedAt:    time.Now().UTC(),
				LogLevel:   event.LogLevelInfo,
				Metadata: &event.ReleaseEventMetadata{
					ReleaseEventCommon: event.ReleaseEventCommon{
						Revision: c.commits[i].Revision,
						Result:   n.Result,
						Error:    n.Result.Error(),
					},
					Spec: event.ReleaseSpec{
						Type:             event.ReleaseImageSpecType,
						ReleaseImageSpec: &spec,
					},
					Cause: n.Spec.Cause,
				},
			})
			eventTypes[event.EventRelease] = true
		case update.Auto:
			spec := n.Spec.Spec.(update.Automated)
			noteEvents = append(noteEvents, event.Event{
				ServiceIDs: n.Result.AffectedResources(),
				Type:       event.EventAutoRelease,
				StartedAt:  started,
				EndedAt:    time.Now().UTC(),
				LogLevel:   event.LogLevelInfo,
				Metadata: &event.AutoReleaseEventMetadata{
					ReleaseEventCommon: event.ReleaseEventCommon{
						Revision: c.commits[i].Revision,
						Result:   n.Result,
						Error:    n.Result.Error(),
					},
					Spec: spec,
				},
			})
			eventTypes[event.EventAutoRelease] = true
		case update.Policy:
			eventTypes[event.EventUpdatePolicy] = true
		default:
			eventTypes[event.NoneOfTheAbove] = true
		}
	}
	return noteEvents, eventTypes, nil
}
func logCommitEvent(el eventLogger, c changeSet, serviceIDs resource.IDSet, started time.Time,
	includesEvents map[string]bool, resourceErrors []event.ResourceError, logger log.Logger) error {
	if len(c.commits) == 0 {
		return nil
	}
	cs := make([]event.Commit, len(c.commits))
	for i, ci := range c.commits {
		cs[i].Revision = ci.Revision
		cs[i].Message = ci.Message
	}
	if err := el.LogEvent(event.Event{
		ServiceIDs: serviceIDs.ToSlice(),
		Type:       event.EventSync,
		StartedAt:  started,
		EndedAt:    started,
		LogLevel:   event.LogLevelInfo,
		Metadata: &event.SyncEventMetadata{
			Commits:     cs,
			InitialSync: c.initialSync,
			Includes:    includesEvents,
			Errors:      resourceErrors,
		},
	}); err != nil {
		logger.Log("err", err)
		return err
	}
	return nil
}