qb/database/worker.go

package database

import (
	"fmt"
	"sync"

	qb "gordenko.dev/dima/qb"
	"gordenko.dev/dima/qb/atree"
	"gordenko.dev/dima/qb/enc"
	"gordenko.dev/dima/qb/proto"
	"gordenko.dev/dima/qb/storage"
	"gordenko.dev/dima/qb/transform"
)

// AddMetric - create metric in map and set Xlock, rwad - append to queue
// DeleteMetric - set Xlock
// DeleteSince - xLock (тупо редкая операция)
// Якийсь прапор поставити для pendingDelete, щоб read операції вставали в чергу

const (
	QueryDone                = 1
	UntilFound               = 2
	UntilNotFound            = 3
	RangeFound               = 15
	NoMeasures               = 16
	NoMetric                 = 4
	MetricDuplicate          = 5
	Succeed                  = 6
	NewPage                  = 7
	ExpiredMeasure           = 8
	NonMonotonicValue        = 9
	CanAppend                = 10
	WrongMetricType          = 11
	NoMeasuresToDelete       = 12
	DeleteFromAtreeNotNeeded = 13
	DeleteFromAtreeRequired  = 14
)

type Worker struct {
	mutex          sync.Mutex
	queue          []any
	signalCh       chan struct{}
	dir            string
	snapshotNumber int
	tmp            []byte
	metrics        map[uint32]*_metric
	saveToStorage  func(any)
	exitCh         chan struct{}
	waitGroup      *sync.WaitGroup
}

type WorkerOptions struct {
	SaveToStorage func(any)
	Metrics       map[uint32]*_metric
	Dir           string
	ExitCh        chan struct{}
	WaitGroup     *sync.WaitGroup
}

func NewWorker(opt WorkerOptions) *Worker {
	return &Worker{
		signalCh:      make(chan struct{}, 1),
		dir:           opt.Dir,
		tmp:           make([]byte, 26),
		metrics:       opt.Metrics,
		saveToStorage: opt.SaveToStorage,
		exitCh:        opt.ExitCh,
		waitGroup:     opt.WaitGroup,
	}
}

func (s *Worker) AddJobToQueue(job any) {
	s.queue = append(s.queue, job)

	select {
	case s.signalCh <- struct{}{}:
	default:
	}
}

func (s *Worker) ReleaseRLock(metricID uint32) {
	s.mutex.Lock()
	//s.rLocksToRelease = append(s.rLocksToRelease, metricID)
	s.mutex.Unlock()

	select {
	case s.signalCh <- struct{}{}:
	default:
	}
}

func (s *Worker) Run() {
	for {
		select {
		case <-s.signalCh:
			s.doWork()

		case <-s.exitCh:
			fmt.Println("worker done")
			s.waitGroup.Done()
			return
		}
	}
}

func (s *Worker) doWork() {

	s.mutex.Lock()
	//rLocksToRelease := s.rLocksToRelease
	queue := s.queue
	//s.rLocksToRelease = nil
	s.queue = nil
	s.mutex.Unlock()

	// for _, metricID := range rLocksToRelease {
	// 	metric, ok := s.metrics[metricID]
	// 	if !ok {
	// 		qb.Abort(qb.NoMetricBug,
	// 			fmt.Errorf("drainQueues: metric %d not found", metricID))
	// 	}

	// 	if metric.XLock {
	// 		qb.Abort(qb.XLockBug,
	// 			fmt.Errorf("drainQueues: xlock is set for the metric %d",
	// 				metricID))
	// 	}

	// 	if metric.RLocks <= 0 {
	// 		qb.Abort(qb.NoRLockBug,
	// 			fmt.Errorf("drainQueues: rlock not set for the metric %d",
	// 				metricID))
	// 	}

	// 	metric.RLocks--

	// 	if len(metric.WaitQueue) > 0 {
	// 		s.processMetricQueue(metricID, metric)
	// 	}
	// }

	for _, untyped := range queue {
		switch req := untyped.(type) {
		case tryAppendMeasuresReq:
			s.tryAppendMeasures(req)

		case storage.Changes:
			s.applyCommits(req) // all metrics only

		case tryListCurrentValuesReq:
			s.tryListCurrentValues(req) // all metrics only

		case tryRangeScanReq:
			s.tryRangeScan(req)

		case tryFullScanReq:
			s.tryFullScan(req)

		case tryAddMetricReq:
			s.tryAddMetric(req)

		case tryDeleteMetricReq:
			s.tryDeleteMetric(req)

		case tryDeleteMeasuresReq:
			s.tryDeleteMeasures(req)

		case tryGetMetricReq:
			s.tryGetMetric(req)

		default:
			qb.Abort(qb.UnknownWorkerQueueItemBug,
				fmt.Errorf("bug: unknown worker queue item type %T", req))
		}
	}
}

// суть у тому що треба запускати запити, пока не зустріну XLock
func (s *Worker) processMetricQueue(metricID uint32, metric *_metric, tmp []byte) {
	if len(metric.WaitQueue) == 0 {
		return
	}

	for _, untyped := range metric.WaitQueue {
		switch req := untyped.(type) {
		case tryRangeScanReq:
			metric.StartRangeScan(req)

		case tryFullScanReq:
			metric.StartFullScan(req)

		case tryGetMetricReq:
			s.tryGetMetric(req)

		case tryAppendMeasuresReq:
			metric.StartAppendMeasures(req, tmp, s.saveToStorage)

		case tryDeleteMetricReq:
			s.startDeleteMetric(metric, req)

		case tryDeleteMeasuresReq:
			s.startDeleteMeasures(metric, req)

		default:
			qb.Abort(qb.UnknownMetricWaitQueueItemBug,
				fmt.Errorf("bug: unknown metric wait queue item type %T", req))
		}
	}
}

type tryAddMetricReq struct {
	MetricID uint32
	ResultCh chan byte
}

func (s *Worker) tryAddMetric(req tryAddMetricReq) {
	_, ok := s.metrics[req.MetricID]
	if ok {
		req.ResultCh <- MetricDuplicate
		return
	}
	fmt.Println("metric not found. Success")
	req.ResultCh <- Succeed // new

	// lockEntry, ok := s.metricLockEntries[req.MetricID]
	// if ok {
	// 	lockEntry.WaitQueue = append(lockEntry.WaitQueue, req)
	// } else {
	// 	s.metricLockEntries[req.MetricID] = &metricLockEntry{
	// 		XLock: true,
	// 	}
	// 	req.ResultCh <- Succeed
	// }
}

func (s *Worker) processTryAddMetricReqsImmediatelyAfterDelete(reqs []tryAddMetricReq) {
	if len(reqs) == 0 {
		return
	}
	var (
		req       = reqs[0]
		waitQueue []any
	)
	if len(reqs) > 1 {
		for _, req := range reqs[1:] {
			waitQueue = append(waitQueue, req)
		}
	}
	// FIX
	// s.metricLockEntries[req.MetricID] = &metricLockEntry{
	// 	XLock:     true,
	// 	WaitQueue: waitQueue,
	// }
	req.ResultCh <- Succeed
}

type tryGetMetricReq struct {
	MetricID uint32
	ResultCh chan Metric
}

func (s *Worker) tryGetMetric(req tryGetMetricReq) {
	metric, ok := s.metrics[req.MetricID]
	if ok {
		req.ResultCh <- Metric{
			ResultCode: Succeed,
			MetricType: metric.metricType,
			FracDigits: metric.fracDigits,
		}
	} else {
		req.ResultCh <- Metric{
			ResultCode: NoMetric,
		}
	}
}

type tryDeleteMetricReq struct {
	MetricID uint32
	ResultCh chan tryDeleteMetricResult
}

func (s *Worker) tryDeleteMetric(req tryDeleteMetricReq) {
	// FIX
	// metric, ok := s.metrics[req.MetricID]
	// if !ok {
	// 	req.ResultCh <- tryDeleteMetricResult{
	// 		ResultCode: NoMetric,
	// 	}
	// 	return
	// }

	// lockEntry, ok := s.metricLockEntries[req.MetricID]
	// if ok {
	// 	lockEntry.WaitQueue = append(lockEntry.WaitQueue, req)
	// } else {
	// 	s.startDeleteMetric(metric, req)
	// }
}

func (s *Worker) startDeleteMetric(metric *_metric, req tryDeleteMetricReq) {
	// FIX
	// s.metricLockEntries[req.MetricID] = &metricLockEntry{
	// 	XLock: true,
	// }
	// req.ResultCh <- tryDeleteMetricResult{
	// 	ResultCode: Succeed,
	// 	RootPageNo: metric.RootPageNo,
	// }
}

type tryDeleteMeasuresReq struct {
	MetricID uint32
	Since    uint32
	ResultCh chan tryDeleteMeasuresResult
}

func (s *Worker) tryDeleteMeasures(req tryDeleteMeasuresReq) {
	// FIX
	// metric, ok := s.metrics[req.MetricID]
	// if !ok {
	// 	req.ResultCh <- tryDeleteMeasuresResult{
	// 		ResultCode: NoMetric,
	// 	}
	// 	return
	// }

	// if metric.Since == 0 || (req.Since > 0 && metric.Until < req.Since) {
	// 	req.ResultCh <- tryDeleteMeasuresResult{
	// 		ResultCode: NoMeasuresToDelete,
	// 	}
	// }

	// lockEntry, ok := s.metricLockEntries[req.MetricID]
	// if ok {
	// 	lockEntry.WaitQueue = append(lockEntry.WaitQueue, req)
	// } else {
	// 	s.startDeleteMeasures(metric, req)
	// }
}

func (s *Worker) startDeleteMeasures(metric *_metric, req tryDeleteMeasuresReq) {
	// FIX
	// s.metricLockEntries[req.MetricID] = &metricLockEntry{
	// 	XLock: true,
	// }

	// if metric.RootPageNo > 0 {
	// 	req.ResultCh <- tryDeleteMeasuresResult{
	// 		ResultCode: DeleteFromAtreeRequired,
	// 		RootPageNo: metric.RootPageNo,
	// 	}
	// } else {
	// 	req.ResultCh <- tryDeleteMeasuresResult{
	// 		ResultCode: DeleteFromAtreeNotNeeded,
	// 	}
	// }
}

func (s *Worker) onMeasuresAppendCommited(rec storage.MeasuresAppendCommited) {
	metric, ok := s.metrics[rec.MetricID]
	if !ok {
		qb.Abort(qb.NoMetricBug,
			fmt.Errorf("finAppendMeasures: metric %d not found",
				rec.MetricID))
	}
	metric.OnMeasuresAppendCommited(rec)
}

func (s *Worker) onMeasuresAppendWithGrowCommited(rec storage.MeasuresAppendWithGrowCommited) {
	metric, ok := s.metrics[rec.MetricID]
	if !ok {
		qb.Abort(qb.NoMetricBug,
			fmt.Errorf("finAppendMeasures: metric %d not found",
				rec.MetricID))
	}
	metric.OnMeasuresAppendWithGrowCommited(rec)
}

type tryAppendMeasuresReq struct {
	MetricID uint32
	Measures []proto.Measure
	ResultCh chan tryAppendMeasuresResult
}

func (s *Worker) tryAppendMeasures(req tryAppendMeasuresReq) {
	metric, ok := s.metrics[req.MetricID]
	if !ok {
		req.ResultCh <- tryAppendMeasuresResult{
			ResultCode: NoMetric,
		}
		return
	}
	if metric.XLock {
		metric.WaitQueue = append(metric.WaitQueue, req)
		return
	}
	metric.StartAppendMeasures(req, s.tmp, s.saveToStorage)
}

type tryRangeScanReq struct {
	MetricID       uint32
	Since          uint32
	Until          uint32
	MetricType     qb.MetricType
	ResponseWriter atree.WorkerMeasureConsumer
	ResultCh       chan rangeScanResult
}

func (s *Worker) tryRangeScan(req tryRangeScanReq) {
	metric, ok := s.metrics[req.MetricID]
	if !ok {
		req.ResultCh <- rangeScanResult{
			ResultCode: NoMetric,
		}
		return
	}
	if metric.metricType != req.MetricType {
		req.ResultCh <- rangeScanResult{
			ResultCode: WrongMetricType,
		}
		return
	}

	if metric.XLock {
		metric.WaitQueue = append(metric.WaitQueue, req)
		return
	}

	metric.StartRangeScan(req)
}

type tryFullScanReq struct {
	MetricID       uint32
	MetricType     qb.MetricType
	ResponseWriter atree.WorkerMeasureConsumer
	ResultCh       chan fullScanResult
}

func (s *Worker) tryFullScan(req tryFullScanReq) {
	metric, ok := s.metrics[req.MetricID]
	if !ok {
		req.ResultCh <- fullScanResult{
			ResultCode: NoMetric,
		}
		return
	}
	if metric.metricType != req.MetricType {
		req.ResultCh <- fullScanResult{
			ResultCode: WrongMetricType,
		}
		return
	}

	if metric.XLock {
		metric.WaitQueue = append(metric.WaitQueue, req)
		return
	}
	metric.StartFullScan(req)
}

type tryListCurrentValuesReq struct {
	MetricIDs      []uint32
	ResponseWriter *transform.CurrentValueWriter
	ResultCh       chan struct{}
}

func (s *Worker) tryListCurrentValues(req tryListCurrentValuesReq) {
	for _, metricID := range req.MetricIDs {
		metric, ok := s.metrics[metricID]
		if ok {
			req.ResponseWriter.BufferValue(transform.CurrentValue{
				MetricID:  metricID,
				Timestamp: metric.LastTimestamp(),
				Value:     metric.LastValue(),
			})
		}
	}
	req.ResultCh <- struct{}{}
}

///////////////////////////////////////////////////////

func (s *Worker) applyCommits(req storage.Changes) {
	for _, untyped := range req.Commits {
		switch rec := untyped.(type) {
		case storage.MetricAddCommited:
			s.onMetricAddCommited(rec)

		case storage.MetricDeleteCommited:
			s.onMetricDeleteCommited(rec)

		case storage.MeasuresAppendCommited:
			s.onMeasuresAppendCommited(rec)

		case storage.MeasuresAppendWithGrowCommited:
			s.onMeasuresAppendWithGrowCommited(rec)

		case storage.MeasuresDeleteCommited:
			s.onMeasuresDeleteCommited(rec)
		}
	}

	if req.SnapshotNumberCh != nil {
		s.snapshotNumber++
		err := writeSnapshot(writeSnapshotIn{
			SnapshotNumber:        s.snapshotNumber,
			Dir:                   s.dir,
			WriteBufferSize:       4 * 1024 * 1024, // 1mb
			Metrics:               s.metrics,
			FrozenIndexPagesCount: req.FrozenIndexPagesCount,
			IndexPageNumbers:      req.IndexPageNumbers,
			FrozenDataPagesCount:  req.FrozenDataPagesCount,
			DataPageNumbers:       req.DataPageNumbers,
		})
		if err != nil {
			qb.Abort(qb.WriteSnapshotFailed, err)
		}

		req.SnapshotNumberCh <- s.snapshotNumber
	}
}

func (s *Worker) onMetricAddCommited(rec storage.MetricAddCommited) {
	// fix lock
	_, ok := s.metrics[rec.MetricID]
	if ok {
		qb.Abort(qb.MetricAddedBug,
			fmt.Errorf("addMetric: metric %d already added",
				rec.MetricID))
	}

	// if !lockEntry.XLock {
	// 	qb.Abort(qb.NoXLockBug,
	// 		fmt.Errorf("addMetric: xlock not set for the metric %d",
	// 			rec.MetricID))
	// }

	// lockEntry.XLock = false
	// delete(s.metricLockEntries, rec.MetricID)
}

func (s *Worker) addMetric(rec storage.MetricAddRecord) {
	var (
		buf     = make([]byte, storage.DataPageSize)
		databuf = buf[:storage.DataPagePayloadSize]
	)
	s.metrics[rec.MetricID] = &_metric{
		metricType: rec.MetricType,
		fracDigits: rec.FracDigits,
		buffer:     buf,
		timestamps: enc.NewTimeDeltaCompressor(databuf, 0),
		values:     enc.NewValueDeltaCompressor(rec.MetricType, rec.FracDigits, databuf, 0),
	}
}

func (s *Worker) onMetricDeleteCommited(rec storage.MetricDeleteCommited) {
	metric, ok := s.metrics[rec.MetricID]
	if !ok {
		qb.Abort(qb.NoMetricBug,
			fmt.Errorf("deleteMetric: metric %d not found",
				rec.MetricID))
	}

	if !metric.XLock {
		qb.Abort(qb.NoXLockBug,
			fmt.Errorf("deleteMetric: xlock not set for the metric %d",
				rec.MetricID))
	}

	var addMetricReqs []tryAddMetricReq

	if len(metric.WaitQueue) > 0 {
		for _, untyped := range metric.WaitQueue {
			switch req := untyped.(type) {
			// case tryAppendMeasureReq:
			// 	req.ResultCh <- tryAppendMeasureResult{
			// 		MetricID:   req.MetricID,
			// 		ResultCode: NoMetric,
			// 	}

			case tryRangeScanReq:
				req.ResultCh <- rangeScanResult{
					ResultCode: NoMetric,
				}

			case tryFullScanReq:
				req.ResultCh <- fullScanResult{
					ResultCode: NoMetric,
				}

			case tryAddMetricReq:
				addMetricReqs = append(addMetricReqs, req)

			case tryDeleteMetricReq:
				req.ResultCh <- tryDeleteMetricResult{
					ResultCode: NoMetric,
				}

			case tryDeleteMeasuresReq:
				req.ResultCh <- tryDeleteMeasuresResult{
					ResultCode: NoMetric,
				}

			case tryGetMetricReq:
				req.ResultCh <- Metric{
					ResultCode: NoMetric,
				}

			default:
				qb.Abort(qb.UnknownMetricWaitQueueItemBug,
					fmt.Errorf("bug: unknown metric wait queue item type %T", req))
			}
		}
	}
	delete(s.metrics, rec.MetricID)

	// ADD in storage
	// if len(rec.FreePageNumbers) > 0 {
	// 	s.freeList.AddPages(rec.FreePageNumbers)
	// }

	if len(addMetricReqs) > 0 {
		s.processTryAddMetricReqsImmediatelyAfterDelete(addMetricReqs)
	}
}

func (s *Worker) onMeasuresDeleteCommited(rec storage.MeasuresDeleteCommited) {
	metric, ok := s.metrics[rec.MetricID]
	if !ok {
		qb.Abort(qb.NoMetricBug,
			fmt.Errorf("deleteMeasures: metric %d not found",
				rec.MetricID))
	}

	if !metric.XLock {
		qb.Abort(qb.NoXLockBug,
			fmt.Errorf("deleteMeasures: xlock not set for the metric %d",
				rec.MetricID))
	}
	metric.OnMeasuresDeleteCommited(rec)
	metric.XLock = false
	// FIX add in storage
	// if len(rec.FreePageNumbers) > 0 {
	// 	s.freeList.AddPages(rec.FreePageNumbers)
	// }
	//s.doAfterReleaseXLock(rec.MetricID, metric)
}

// func (s *Database) doAfterReleaseXLock(metricID uint32, metric *_metric) {
// 	if len(metric.WaitQueue) > 0 {
// 		s.processMetricQueue(metricID, metric)
// 	}
// }

func (s *Worker) recovery() {
	// FIX
	// advisor, err := recovery.NewRecoveryAdvisor(recovery.RecoveryAdvisorOptions{
	// 	Dir:            s.dir,
	// 	VerifySnapshot: s.verifySnapshot,
	// })
	// if err != nil {
	// 	panic(err)
	// }

	// recipe, err := advisor.GetRecipe()
	// if err != nil {
	// 	qb.Abort(qb.GetRecoveryRecipeFailed, err)
	// }

	// var logNumber int

	// if recipe != nil {
	// 	if recipe.Snapshot != "" {
	// 		err = s.loadSnapshot(recipe.Snapshot)
	// 		if err != nil {
	// 			qb.Abort(qb.LoadSnapshotFailed, err)
	// 		}
	// 	}
	// 	for _, changesFileName := range recipe.Changes {
	// 		err = s.replayChanges(changesFileName)
	// 		if err != nil {
	// 			qb.Abort(qb.ReplayChangesFailed, err)
	// 		}
	// 	}
	// 	logNumber = recipe.LogNumber
	// }

	// s.storage, err = storage.NewWriter(storage.WriterOptions{
	// 	Dir:                 s.dir,
	// 	LogNumber:           logNumber,
	// 	AppendToWorkerQueue: s.appendJobToWorkerQueue,
	// 	FreeList:            s.freeList,
	// 	Atree:               s.atree,
	// 	ExitCh:              s.exitCh,
	// 	WaitGroup:           s.waitGroup,
	// })
	// if err != nil {
	// 	qb.Abort(qb.CreateChangesWriterFailed, err)

	// }
	// go s.storage.Run()

	// // fileNames, err := s.searchREDOFiles()
	// // if err != nil {
	// // 	qb.Abort(qb.SearchREDOFilesFailed, err)
	// // }

	// // if len(fileNames) > 0 {
	// // 	for _, fileName := range fileNames {
	// // 		err = s.replayREDOFile(fileName)
	// // 		if err != nil {
	// // 			qb.Abort(qb.ReplayREDOFileFailed, err)
	// // 		}
	// // 	}

	// // 	for _, fileName := range fileNames {
	// // 		err = os.Remove(fileName)
	// // 		if err != nil {
	// // 			qb.Abort(qb.RemoveREDOFileFailed, err)
	// // 		}
	// // 	}
	// // }

	// if recipe != nil {
	// 	if recipe.CompleteSnapshot {
	// 		err = s.dumpSnapshot(logNumber)
	// 		if err != nil {
	// 			qb.Abort(qb.DumpSnapshotFailed, err)
	// 		}
	// 	}

	// 	for _, fileName := range recipe.ToDelete {
	// 		err = os.Remove(fileName)
	// 		if err != nil {
	// 			qb.Abort(qb.RemoveRecipeFileFailed, err)
	// 		}
	// 	}
	// }
}