api/services/changes.go

package services

import (
	"context"
	"git.aiterp.net/rpdata/api/internal/auth"
	"git.aiterp.net/rpdata/api/internal/notifier"
	"git.aiterp.net/rpdata/api/models"
	"git.aiterp.net/rpdata/api/repositories"
	"log"
	"sync"
	"time"
)

type ChangeService struct {
	changes repositories.ChangeRepository

	mutex       sync.RWMutex
	buffer      []models.Change
	offset      uint64
	notifier    notifier.Notifier
	submitQueue chan *models.Change
	loopStarted bool
}

func (s *ChangeService) Find(ctx context.Context, id string) (*models.Change, error) {
	return s.changes.Find(ctx, id)
}

func (s *ChangeService) List(ctx context.Context, filter models.ChangeFilter) ([]*models.Change, error) {
	return s.changes.List(ctx, filter)
}

func (s *ChangeService) Submit(ctx context.Context, model models.ChangeModel, op string, listed bool, keys []models.ChangeKey, objects ...interface{}) {
	token := auth.TokenFromContext(ctx)
	if token == nil {
		panic("no token!")
	}

	change := &models.Change{
		Model:  model,
		Op:     op,
		Author: token.UserID,
		Listed: listed,
		Keys:   keys,
		Date:   time.Now(),
	}

	for _, obj := range objects {
		if !change.AddObject(obj) {
			log.Printf("Cannot add object of type %T to change", obj)
		}
	}

	s.mutex.Lock()
	if !s.loopStarted {
		s.loopStarted = true
		s.submitQueue = make(chan *models.Change, 64)
		go s.loop()
	}
	s.mutex.Unlock()

	s.submitQueue <- change
}

func (s *ChangeService) Subscribe(ctx context.Context, filter models.ChangeFilter) <-chan *models.Change {
	channel := make(chan *models.Change)

	go func() {
		defer close(channel)

		s.mutex.RLock()
		pos := s.offset + uint64(len(s.buffer))
		slice := make([]models.Change, 32)
		s.mutex.RUnlock()

		count := 0

		for {
			s.mutex.RLock()
			nextPos := s.offset + uint64(len(s.buffer))
			length := nextPos - pos
			if length > 0 {
				index := pos - s.offset

				pos = nextPos
				copy(slice, s.buffer[index:])
			}
			ch := s.notifier.C()
			s.mutex.RUnlock()

			for _, change := range slice[:length] {
				if change.PassesFilter(filter) {
					channel <- &change

					count++
					if filter.Limit != nil && count == *filter.Limit {
						return
					}
				}
			}

			select {
			case <-ch:
			case <-ctx.Done():
				return
			}
		}
	}()

	return channel
}

func (s *ChangeService) loop() {
	for change := range s.submitQueue {
		timeout, cancel := context.WithTimeout(context.Background(), time.Second*15)

		change, err := s.changes.Insert(timeout, *change)
		if err != nil {
			log.Println("Failed to insert change:")
		} else {
			log.Println("Change", change.ID, "inserted.")
		}

		s.mutex.Lock()
		s.buffer = append(s.buffer, *change)
		if len(s.buffer) > 16 {
			copy(s.buffer, s.buffer[8:])
			s.buffer = s.buffer[:len(s.buffer)-8]
			s.offset += 8
		}
		s.mutex.Unlock()
		s.notifier.Broadcast()

		cancel()
	}
}