indigo-ci / backfill / memstore.go

package backfill

import (
	"context"
	"fmt"
	"sync"
	"time"

	"github.com/bluesky-social/indigo/repomgr"
	"github.com/ipfs/go-cid"
)

// A BufferedOp is an operation buffered while a repo is being backfilled.
type BufferedOp struct {
	// Kind describes the type of operation.
	Kind repomgr.EventKind
	// Path contains the path the operation applies to.
	Path string
	// Record contains the serialized record for create and update operations.
	Record *[]byte
	// Cid is the CID of the record.
	Cid *cid.Cid
}

type opSet struct {
	since *string
	rev   string
	ops   []*BufferedOp
}

type Memjob struct {
	repo        string
	state       string
	rev         string
	lk          sync.Mutex
	bufferedOps []*opSet

	createdAt time.Time
	updatedAt time.Time
}

// Memstore is a simple in-memory implementation of the Backfill Store interface
type Memstore struct {
	lk   sync.RWMutex
	jobs map[string]*Memjob
}

func NewMemstore() *Memstore {
	return &Memstore{
		jobs: make(map[string]*Memjob),
	}
}

func (s *Memstore) EnqueueJob(repo string) error {
	s.lk.Lock()
	defer s.lk.Unlock()

	if _, ok := s.jobs[repo]; ok {
		return fmt.Errorf("job already exists for repo %s", repo)
	}

	j := &Memjob{
		repo:      repo,
		createdAt: time.Now(),
		updatedAt: time.Now(),
		state:     StateEnqueued,
	}
	s.jobs[repo] = j
	return nil
}

func (s *Memstore) EnqueueJobWithState(repo, state string) error {
	s.lk.Lock()
	defer s.lk.Unlock()

	if _, ok := s.jobs[repo]; ok {
		return fmt.Errorf("job already exists for repo %s", repo)
	}

	j := &Memjob{
		repo:      repo,
		createdAt: time.Now(),
		updatedAt: time.Now(),
		state:     state,
	}
	s.jobs[repo] = j
	return nil
}

func (s *Memstore) BufferOp(ctx context.Context, repo string, since *string, rev string, kind repomgr.EventKind, path string, rec *[]byte, cid *cid.Cid) (bool, error) {
	s.lk.Lock()

	// If the job doesn't exist, we can't buffer an op for it
	j, ok := s.jobs[repo]
	s.lk.Unlock()
	if !ok {
		return false, ErrJobNotFound
	}

	j.lk.Lock()
	defer j.lk.Unlock()

	switch j.state {
	case StateComplete:
		return false, ErrJobComplete
	case StateInProgress:
	// keep going and buffer the op
	default:
		return false, nil
	}

	j.bufferedOps = append(j.bufferedOps, &opSet{
		since: since,
		rev:   rev,
		ops: []*BufferedOp{&BufferedOp{
			Path:   path,
			Kind:   kind,
			Record: rec,
			Cid:    cid,
		}},
	})
	j.updatedAt = time.Now()
	return true, nil
}

func (j *Memjob) BufferOps(ctx context.Context, since *string, rev string, ops []*BufferedOp) (bool, error) {
	j.lk.Lock()
	defer j.lk.Unlock()

	switch j.state {
	case StateComplete:
		return false, ErrJobComplete
	case StateInProgress:
	// keep going and buffer the op
	default:
		return false, nil
	}

	j.bufferedOps = append(j.bufferedOps, &opSet{
		since: since,
		rev:   rev,
		ops:   ops,
	})
	j.updatedAt = time.Now()
	return true, nil
}

func (s *Memstore) GetJob(ctx context.Context, repo string) (Job, error) {
	s.lk.RLock()
	defer s.lk.RUnlock()

	j, ok := s.jobs[repo]
	if !ok || j == nil {
		return nil, nil
	}
	return j, nil
}

func (s *Memstore) GetNextEnqueuedJob(ctx context.Context) (Job, error) {
	s.lk.RLock()
	defer s.lk.RUnlock()

	for _, j := range s.jobs {
		if j.State() == StateEnqueued {
			return j, nil
		}
	}
	return nil, nil
}

func (s *Memstore) PurgeRepo(ctx context.Context, repo string) error {
	s.lk.RLock()
	defer s.lk.RUnlock()

	delete(s.jobs, repo)
	return nil
}

func (j *Memjob) Repo() string {
	return j.repo
}

func (j *Memjob) State() string {
	j.lk.Lock()
	defer j.lk.Unlock()

	return j.state
}

func (j *Memjob) SetState(ctx context.Context, state string) error {
	j.lk.Lock()
	defer j.lk.Unlock()

	j.state = state
	j.updatedAt = time.Now()
	return nil
}

func (j *Memjob) Rev() string {
	return j.rev
}

func (j *Memjob) SetRev(ctx context.Context, rev string) error {
	j.rev = rev
	return nil
}

func (j *Memjob) FlushBufferedOps(ctx context.Context, fn func(kind repomgr.EventKind, rev, path string, rec *[]byte, cid *cid.Cid) error) error {
	panic("TODO: copy what we end up doing from the gormstore")
	/*
		j.lk.Lock()
		defer j.lk.Unlock()

		for _, opset := range j.bufferedOps {
			for _, op := range opset.ops {
				if err := fn(op.Kind, op.Path, op.Record, op.Cid); err != nil {
					return err
				}
			}
		}

		j.bufferedOps = map[string][]*BufferedOp{}
		j.state = StateComplete

		return nil
	*/
}

func (j *Memjob) ClearBufferedOps(ctx context.Context) error {
	j.lk.Lock()
	defer j.lk.Unlock()

	j.bufferedOps = []*opSet{}
	j.updatedAt = time.Now()
	return nil
}

func (j *Memjob) RetryCount() int {
	j.lk.Lock()
	defer j.lk.Unlock()
	return 0
}