manager.go

package manager

import (
	"fmt"
	"math"
	"strings"
	"time"

	"github.com/lbryio/transcoder/library"
	db "github.com/lbryio/transcoder/library/db"
	"github.com/lbryio/transcoder/pkg/mfr"
	"github.com/lbryio/transcoder/pkg/resolve"

	"github.com/karlseguin/ccache/v2"
)

const (
	videoPlaylistPath      = "."
	channelURIPrefix       = "lbry://"
	level5SupportThreshold = 1000
)

var (
	cacheSize = int64(math.Pow(1024, 4))
)

type TranscodingRequest struct {
	resolve.ResolvedStream
	queue *mfr.Queue
}

type VideoLibrary interface {
	GetURL(sdHash string) (string, error)
	GetAllChannels() ([]db.Channel, error)
	// Add(params video.AddParams) (*video.Video, error)
	// AddLocalStream(url, channel string, ls storage.LocalStream) (*video.Video, error)
	// AddRemoteStream(storage.RemoteStream) (*video.Video, error)
}

type VideoManager struct {
	lib      *library.Library
	pool     *Pool
	cache    *ccache.Cache
	channels *channelList
}

// NewManager creates a video library manager with a pool for future transcoding requests.
func NewManager(lib *library.Library, minHits int) *VideoManager {
	m := &VideoManager{
		lib:      lib,
		pool:     NewPool(),
		channels: newChannelList(),
		cache: ccache.New(ccache.
			Configure().
			MaxSize(cacheSize)),
	}

	channels, err := lib.GetAllChannels()
	if err != nil {
		logger.Error("error loading channels", "err", err)
	}
	m.channels.Load(channels)
	logger.Infow("loaded channels", "count", len(channels))
	go m.channels.StartLoadingChannels(lib)

	m.pool.AddQueue("priority", 0, func(key string, value interface{}, queue *mfr.Queue) bool {
		r := value.(*TranscodingRequest)
		if m.channels.GetPriority(r) != db.ChannelPriorityHigh {
			return false
		}
		logger.Infow("accepted for 'priority' queue", "uri", r.URI)
		r.queue = queue
		queue.Hit(key, r)
		return true
	})

	m.pool.AddQueue("enabled", 0, func(key string, value interface{}, queue *mfr.Queue) bool {
		r := value.(*TranscodingRequest)
		if m.channels.GetPriority(r) != db.ChannelPriorityNormal {
			return false
		}
		logger.Infow("accepted for 'priority' queue", "uri", r.URI)
		r.queue = queue
		queue.Hit(key, r)
		return true
	})

	m.pool.AddQueue("level5", 0, func(key string, value interface{}, queue *mfr.Queue) bool {
		r := value.(*TranscodingRequest)
		s := r.ChannelSupportAmount
		if level5SupportThreshold > s {
			return false
		}
		r.ChannelSupportAmount = 0
		logger.Debugw("accepted for 'level5' queue", "uri", r.URI, "support_amount", r.ChannelSupportAmount)
		r.queue = queue
		queue.Hit(key, r)
		return true
	})

	m.pool.AddQueue("common", uint(minHits), func(key string, value interface{}, queue *mfr.Queue) bool {
		r := value.(*TranscodingRequest)
		if m.channels.GetPriority(r) == db.ChannelPriorityDisabled {
			return false
		}
		logger.Debugw("accepted for 'common' queue", "uri", r.URI, "support_amount", r.ChannelSupportAmount)
		r.queue = queue
		queue.Hit(key, r)
		return true
	})

	go m.pool.Start()

	return m
}

func (m *VideoManager) Pool() *Pool {
	return m.pool
}

func (m *VideoManager) RequestStatus(sdHash string) int {
	for _, l := range m.pool.levels {
		if _, status := l.queue.Get(sdHash); status != mfr.StatusNone {
			return status
		}
	}
	return mfr.StatusNone
}

func (m *VideoManager) Library() *library.Library {
	return m.lib
}

// Video checks if video exists in the library or waiting in one of the queues.
// If neither, it adds claim to the pool for later processing.
func (m *VideoManager) Video(uri string) (string, error) {
	uri = strings.TrimPrefix(uri, "lbry://")
	tr, err := m.ResolveStream(uri)
	if err != nil {
		return "", err
	}

	if m.channels.GetPriority(tr) == db.ChannelPriorityDisabled {
		return "", resolve.ErrTranscodingForbidden
	}

	vloc, err := m.lib.GetVideoURL(tr.SDHash)
	if err != nil {
		return "", m.pool.Admit(tr.SDHash, tr)
	}

	return vloc, nil
}

// Requests returns next transcoding request to be processed. It polls all queues in the pool evenly.
func (m *VideoManager) Requests() <-chan *TranscodingRequest {
	out := make(chan *TranscodingRequest)
	go func() {
		for next := range m.pool.Out() {
			if next == nil {
				continue
			}

			r := next.Value.(*TranscodingRequest)
			logger.Infow("pulling out next request", "uri", r.URI, "hits", next.Hits())
			out <- r
		}
	}()
	return out
}

func (m *VideoManager) ResolveStream(uri string) (*TranscodingRequest, error) {
	item, err := m.cache.Fetch(fmt.Sprintf("claim:%v", uri), 300*time.Second, func() (interface{}, error) {
		return resolve.ResolveStream(uri)
	})
	if err != nil {
		return nil, err
	}
	rs := item.Value().(*resolve.ResolvedStream)
	return &TranscodingRequest{ResolvedStream: *rs}, nil
}

func (r *TranscodingRequest) Release() {
	if r.queue == nil {
		return
	}
	logger.Infow("transcoding request released", "lbry_url", r.URI)
	r.queue.Release(r.URI)
}

func (r *TranscodingRequest) Reject() {
	if r.queue == nil {
		return
	}
	logger.Infow("transcoding request rejected", "lbry_url", r.URI)
	r.queue.Done(r.URI)
}

func (r *TranscodingRequest) Complete() {
	if r.queue == nil {
		return
	}
	logger.Infow("transcoding request completed", "lbry_url", r.URI)
	r.queue.Done(r.URI)
}