qb/database/api.go

package database

import (
	"errors"
	"fmt"
	"io"
	"net"

	qb "gordenko.dev/dima/qb"
	"gordenko.dev/dima/qb/atree"
	"gordenko.dev/dima/qb/bin"
	"gordenko.dev/dima/qb/bufreader"
	"gordenko.dev/dima/qb/proto"
	"gordenko.dev/dima/qb/transform"
	"gordenko.dev/dima/qb/txlog"
)

var (
	ErrNoValueBug         = errors.New("has timestamp but no value")
	ErrWrongResultCodeBug = errors.New("bug: wrong result code")

	successMsg = []byte{
		proto.RespSuccess,
	}
)

func reply(conn io.Writer, errcode uint16) {
	var answer []byte
	if errcode == 0 {
		answer = successMsg
	} else {
		answer = []byte{
			proto.RespError,
			0, 0,
		}
		bin.PutUint16(answer[1:], errcode)
	}

	_, err := conn.Write(answer)
	if err != nil {
		return
	}
}

func (s *Database) handleTCPConn(conn net.Conn) {
	defer conn.Close()

	r := bufreader.New(conn, 128)

	for {
		err := s.processRequest(conn, r)
		if err != nil {
			if err != io.EOF {
				s.logger.Println(err)
			}
			return
		}
	}
}

func (s *Database) processRequest(conn net.Conn, r *bufreader.BufferedReader) (err error) {
	messageType, err := r.ReadByte()
	if err != nil {
		if err != io.EOF {
			return fmt.Errorf("read messageType: %s", err)
		} else {
			return err
		}
	}

	switch messageType {
	case proto.TypeGetMetric:
		req, err := proto.ReadGetMetricReq(r)
		if err != nil {
			return fmt.Errorf("proto.ReadGetMetricReq: %s", err)
		}
		if err = s.GetMetric(conn, req); err != nil {
			return fmt.Errorf("GetMetric: %s", err)
		}

	case proto.TypeAddMetric:
		req, err := proto.ReadAddMetricReq(r)
		if err != nil {
			return fmt.Errorf("proto.ReadAddMetricReq: %s", err)
		}
		reply(conn, s.AddMetric(req))

	case proto.TypeDeleteMetric:
		req, err := proto.ReadDeleteMetricReq(r)
		if err != nil {
			return fmt.Errorf("proto.ReadDeleteMetricReq: %s", err)
		}
		reply(conn, s.DeleteMetric(req))

	// case proto.TypeAppendMeasure:
	// 	req, err := proto.ReadAppendMeasureReq(r)
	// 	if err != nil {
	// 		return fmt.Errorf("proto.ReadAppendMeasureReq: %s", err)
	// 	}
	// 	//fmt.Println("append measure", req.MetricID, conn.RemoteAddr().String())
	// 	reply(conn, s.AppendMeasure(req))

	case proto.TypeAppendMeasures:
		req, err := proto.ReadAppendMeasuresReq(r)
		if err != nil {
			return fmt.Errorf("proto.ReadAppendMeasuresReq: %s", err)
		}
		//fmt.Println("append measure", req.MetricID, conn.RemoteAddr().String())
		reply(conn, s.AppendMeasures(req))

	case proto.TypeListInstantMeasures:
		req, err := proto.ReadListInstantMeasuresReq(r)
		if err != nil {
			return fmt.Errorf("proto.ReadListInstantMeasuresReq: %s", err)
		}
		if err = s.ListInstantMeasures(conn, req); err != nil {
			return fmt.Errorf("ListInstantMeasures: %s", err)
		}

	case proto.TypeListCumulativeMeasures:
		req, err := proto.ReadListCumulativeMeasuresReq(r)
		if err != nil {
			return fmt.Errorf("proto.ReadListCumulativeMeasuresReq: %s", err)
		}
		if err = s.ListCumulativeMeasures(conn, req); err != nil {
			return fmt.Errorf("ListCumulativeMeasures: %s", err)
		}

	case proto.TypeListInstantPeriods:
		req, err := proto.ReadListInstantPeriodsReq(r)
		if err != nil {
			return fmt.Errorf("proto.ReadListInstantPeriodsReq: %s", err)
		}
		if err = s.ListInstantPeriods(conn, req); err != nil {
			return fmt.Errorf("ListInstantPeriods: %s", err)
		}

	case proto.TypeListCumulativePeriods:
		req, err := proto.ReadListCumulativePeriodsReq(r)
		if err != nil {
			return fmt.Errorf("proto.ReadListCumulativePeriodsReq: %s", err)
		}
		if err = s.ListCumulativePeriods(conn, req); err != nil {
			return fmt.Errorf("ListCumulativePeriods: %s", err)
		}

	case proto.TypeListCurrentValues:
		req, err := proto.ReadListCurrentValuesReq(r)
		if err != nil {
			return fmt.Errorf("proto.ListCurrentValuesReq: %s", err)
		}
		if err = s.ListCurrentValues(conn, req); err != nil {
			return fmt.Errorf("ListCurrentValues: %s", err)
		}

	case proto.TypeDeleteMeasures:
		req, err := proto.ReadDeleteMeasuresReq(r)
		if err != nil {
			return fmt.Errorf("proto.ReadDeleteMeasuresReq: %s", err)
		}
		reply(conn, s.DeleteMeasures(req))

	case proto.TypeListAllInstantMeasures:
		req, err := proto.ReadListAllInstantMeasuresReq(r)
		if err != nil {
			return fmt.Errorf("proto.ReadListAllInstantMeasuresReq: %s", err)
		}
		if err = s.ListAllInstantMeasures(conn, req); err != nil {
			return fmt.Errorf("ListAllInstantMeasures: %s", err)
		}

	case proto.TypeListAllCumulativeMeasures:
		req, err := proto.ReadListAllCumulativeMeasuresReq(r)
		if err != nil {
			return fmt.Errorf("proto.ReadListAllCumulativeMeasuresReq: %s", err)
		}

		if err = s.ListAllCumulativeMeasures(conn, req); err != nil {
			return fmt.Errorf("ListAllCumulativeMeasures: %s", err)
		}

	default:
		return fmt.Errorf("unknown messageType: %d", messageType)
	}
	return
}

// API

func (s *Database) AddMetric(req proto.AddMetricReq) uint16 {
	// Валидация
	if req.MetricID == 0 {
		return proto.ErrEmptyMetricID
	}
	if byte(req.FracDigits) > qb.MaxFracDigits {
		return proto.ErrWrongFracDigits
	}
	switch req.MetricType {
	case qb.Cumulative, qb.Instant:
		// ok
	default:
		return proto.ErrWrongMetricType
	}

	resultCh := make(chan byte, 1)

	s.appendJobToWorkerQueue(tryAddMetricReq{
		MetricID: req.MetricID,
		ResultCh: resultCh,
	})

	resultCode := <-resultCh

	switch resultCode {
	case Succeed:
		s.txlog.Append(txlog.AddedMetric{
			MetricID:   req.MetricID,
			MetricType: req.MetricType,
			FracDigits: req.FracDigits,
		})
		//<-waitCh

	case MetricDuplicate:
		return proto.ErrDuplicate

	default:
		qb.Abort(qb.WrongResultCodeBug, ErrWrongResultCodeBug)
	}
	return 0
}

type Metric struct {
	MetricType qb.MetricType
	FracDigits byte
	ResultCode byte
}

func (s *Database) GetMetric(conn io.Writer, req proto.GetMetricReq) error {
	resultCh := make(chan Metric, 1)

	s.appendJobToWorkerQueue(tryGetMetricReq{
		MetricID: req.MetricID,
		ResultCh: resultCh,
	})

	result := <-resultCh

	switch result.ResultCode {
	case Succeed:
		answer := []byte{
			proto.RespValue,
			0, 0, 0, 0, // metricID
			byte(result.MetricType),
			result.FracDigits,
		}
		bin.PutUint32(answer[1:], req.MetricID)

		_, err := conn.Write(answer)
		if err != nil {
			return err
		}

	case NoMetric:
		reply(conn, proto.ErrNoMetric)

	default:
		qb.Abort(qb.WrongResultCodeBug, ErrWrongResultCodeBug)
	}
	return nil
}

type tryDeleteMetricResult struct {
	ResultCode byte
	RootPageNo uint32
}

func (s *Database) DeleteMetric(req proto.DeleteMetricReq) uint16 {
	resultCh := make(chan tryDeleteMetricResult, 1)

	s.appendJobToWorkerQueue(tryDeleteMetricReq{
		MetricID: req.MetricID,
		ResultCh: resultCh,
	})

	result := <-resultCh

	switch result.ResultCode {
	case Succeed:
		var (
			freePageNumbers []uint32
		)
		if result.RootPageNo > 0 {
			var err error
			freePageNumbers, err = s.atree.GetAllPages(result.RootPageNo)
			if err != nil {
				qb.Abort(qb.FailedAtreeRequest, err)
			}
		}
		s.txlog.Append(txlog.DeletedMetric{
			MetricID:        req.MetricID,
			FreePageNumbers: freePageNumbers,
		})
		//<-waitCh

	case NoMetric:
		return proto.ErrNoMetric

	default:
		qb.Abort(qb.WrongResultCodeBug, ErrWrongResultCodeBug)
	}
	return 0
}

type FilledPage struct {
	Since            uint32
	RootPageNo       uint32
	PrevPageNo       uint32
	TimestampsChunks [][]byte
	TimestampsSize   uint16
	ValuesChunks     [][]byte
	ValuesSize       uint16
}

// type tryAppendMeasureResult struct {
// 	MetricID   uint32
// 	Timestamp  uint32
// 	Value      float64
// 	FilledPage *FilledPage
// 	ResultCode byte
// }

// func (s *Database) AppendMeasure(req proto.AppendMeasureReq) uint16 {
// 	resultCh := make(chan tryAppendMeasureResult, 1)

// 	s.appendJobToWorkerQueue(tryAppendMeasureReq{
// 		MetricID:  req.MetricID,
// 		Timestamp: req.Timestamp,
// 		Value:     req.Value,
// 		ResultCh:  resultCh,
// 	})

// 	result := <-resultCh

// 	switch result.ResultCode {
// 	case CanAppend:
// 		waitCh := s.txlog.Append(txlog.AppendedMeasure{
// 			MetricID:  req.MetricID,
// 			Timestamp: req.Timestamp,
// 			Value:     req.Value,
// 		})
// 		<-waitCh

// 	case NewPage:
// 		filled := result.FilledPage

// 		var path atree.PathToDataPage
// 		if filled.RootPageNo > 0 {
// 			path, err := s.atree.FindPathToLastPage(filled.RootPageNo)
// 			if err != nil {
// 				// FIX
// 				qb.Abort(qb.WriteToAtreeFailed, err)
// 			}
// 			// for _, leg := range path.Legs {
// 			// 	pagesToRelease = append(pagesToRelease, leg.PageNo)
// 			// }

// 			if path.LastPageNo != filled.PrevPageNo {
// 				qb.Abort(
// 					qb.WrongPrevPageNo,
// 					fmt.Errorf("bug: last pageNo %d in tree != prev pageNo %d in _metric",
// 						path.LastPageNo, filled.PrevPageNo),
// 				)
// 			}
// 		}

// 		// report, err := s.atree.AppendDataPages(atree.AppendDataPageReq{
// 		// 	MetricID:         req.MetricID,
// 		// 	Timestamp:        req.Timestamp,
// 		// 	Value:            req.Value,
// 		// 	Since:            filled.Since,
// 		// 	RootPageNo:       filled.RootPageNo,
// 		// 	PrevPageNo:       filled.PrevPageNo,
// 		// 	TimestampsChunks: filled.TimestampsChunks,
// 		// 	TimestampsSize:   filled.TimestampsSize,
// 		// 	ValuesChunks:     filled.ValuesChunks,
// 		// 	ValuesSize:       filled.ValuesSize,
// 		// })
// 		// if err != nil {
// 		// 	qb.Abort(qb.WriteToAtreeFailed, err)
// 		// }
// 		// _ = report

// 		//waitCh := s.txlog.WriteAppended(path, report)

// 		waitCh := s.txlog.Append(txlog.AppendedPagesReq{
// 			Legs:       path.Legs,
// 			MetricID:   req.MetricID,
// 			Timestamp:  req.Timestamp,
// 			Value:      req.Value,
// 			LastPageNo: filled.PrevPageNo,
// 			//Pages: ,
// 			TimestampsChunks: filled.TimestampsChunks,
// 			TimestampsSize:   filled.TimestampsSize,
// 			ValuesChunks:     filled.ValuesChunks,
// 			ValuesSize:       filled.ValuesSize,
// 		},
// 		)
// 		<-waitCh

// 	case NoMetric:
// 		return proto.ErrNoMetric

// 	case ExpiredMeasure:
// 		return proto.ErrExpiredMeasure

// 	case NonMonotonicValue:
// 		return proto.ErrNonMonotonicValue

// 	default:
// 		qb.Abort(qb.WrongResultCodeBug, ErrWrongResultCodeBug)
// 	}
// 	return 0
// }

type tryAppendMeasuresResult struct {
	ResultCode byte
	Written    int
	// MetricType    qb.MetricType
	// FracDigits    byte
	// Since         uint32
	// Until         uint32
	// UntilValue    float64
	// RootPageNo    uint32
	// PrevPageNo    uint32
	// TimestampsBuf *conbuf.ContinuousBuffer
	// ValuesBuf     *conbuf.ContinuousBuffer
	// Timestamps    qb.TimestampCompressor
	// Values        qb.ValueCompressor
}

func (s *Database) AppendMeasures(req proto.AppendMeasuresReq) uint16 {
	resultCh := make(chan tryAppendMeasuresResult, 1)

	s.appendJobToWorkerQueue(tryAppendMeasuresReq{
		MetricID: req.MetricID,
		Measures: req.Measures,
		ResultCh: resultCh,
	})

	result := <-resultCh

	switch result.ResultCode {
	case CanAppend:

		// report, err := s.atree.AppendDataPage(atree.AppendDataPageReq{})
		// if err != nil {
		// 	qb.Abort(qb.WriteToAtreeFailed, err)
		// }
		// _ = report

		// if len(toAppendMeasures) > 0 {
		// 	waitCh := s.txlog.WriteAppendMeasures(
		// 		txlog.AppendedMeasures{
		// 			MetricID: req.MetricID,
		// 			Measures: toAppendMeasures,
		// 		},
		// 		false,
		// 	)
		// 	<-waitCh
		// }

	case NoMetric:
		return proto.ErrNoMetric

	default:
		qb.Abort(qb.WrongResultCodeBug, ErrWrongResultCodeBug)
	}
	return 0
}

type tryDeleteMeasuresResult struct {
	ResultCode byte
	RootPageNo uint32
}

func (s *Database) DeleteMeasures(req proto.DeleteMeasuresReq) uint16 {
	resultCh := make(chan tryDeleteMeasuresResult, 1)

	s.appendJobToWorkerQueue(tryDeleteMeasuresReq{
		MetricID: req.MetricID,
		Since:    req.Since,
		ResultCh: resultCh,
	})

	result := <-resultCh

	switch result.ResultCode {
	case NoMeasuresToDelete:
		// ok

	case DeleteFromAtreeNotNeeded:
		// регистрирую удаление в TransactionLog
		s.txlog.Append(txlog.DeletedMeasures{
			MetricID: req.MetricID,
		})
		//<-waitCh

	case DeleteFromAtreeRequired:
		// собираю номера всех data и index страниц метрики (типа запись REDO лога).
		pageNumbers, err := s.atree.GetAllPages(req.MetricID)
		if err != nil {
			qb.Abort(qb.FailedAtreeRequest, err)
		}
		// регистрирую удаление в TransactionLog
		s.txlog.Append(txlog.DeletedMeasures{
			MetricID:        req.MetricID,
			FreePageNumbers: pageNumbers,
		})
		//<-waitCh

	case NoMetric:
		return proto.ErrNoMetric

	default:
		qb.Abort(qb.WrongResultCodeBug, ErrWrongResultCodeBug)
	}
	return 0
}

// SELECT

type fullScanResult struct {
	ResultCode byte
	FracDigits byte
	LastPageNo uint32
}

func (s *Database) ListAllInstantMeasures(conn net.Conn, req proto.ListAllInstantMetricMeasuresReq) error {
	responseWriter := transform.NewInstantMeasureWriter(conn, 0)

	return s.fullScan(fullScanReq{
		MetricID:       req.MetricID,
		MetricType:     qb.Instant,
		Conn:           conn,
		ResponseWriter: responseWriter,
	})
}

func (s *Database) ListAllCumulativeMeasures(conn io.Writer, req proto.ListAllCumulativeMeasuresReq) error {
	responseWriter := transform.NewCumulativeMeasureWriter(conn, 0)

	return s.fullScan(fullScanReq{
		MetricID:       req.MetricID,
		MetricType:     qb.Cumulative,
		Conn:           conn,
		ResponseWriter: responseWriter,
	})
}

func (s *Database) ListInstantMeasures(conn net.Conn, req proto.ListInstantMeasuresReq) error {
	if req.Since > req.Until {
		reply(conn, proto.ErrInvalidRange)
		return nil
	}

	responseWriter := transform.NewInstantMeasureWriter(conn, req.Since)

	return s.rangeScan(rangeScanReq{
		MetricID:       req.MetricID,
		MetricType:     qb.Instant,
		Since:          req.Since,
		Until:          req.Until - 1,
		Conn:           conn,
		ResponseWriter: responseWriter,
	})
}

func (s *Database) ListCumulativeMeasures(conn net.Conn, req proto.ListCumulativeMeasuresReq) error {
	if req.Since > req.Until {
		reply(conn, proto.ErrInvalidRange)
		return nil
	}

	responseWriter := transform.NewCumulativeMeasureWriter(conn, req.Since)

	return s.rangeScan(rangeScanReq{
		MetricID:       req.MetricID,
		MetricType:     qb.Cumulative,
		Since:          req.Since,
		Until:          req.Until - 1,
		Conn:           conn,
		ResponseWriter: responseWriter,
	})
}

type rangeScanResult struct {
	ResultCode byte
	FracDigits byte
	RootPageNo uint32
	LastPageNo uint32
}

func (s *Database) ListInstantPeriods(conn net.Conn, req proto.ListInstantPeriodsReq) error {
	since, until := timeBoundsOfAggregation(req.Since, req.Until, req.GroupBy, req.FirstHourOfDay)
	if since.After(until) {
		reply(conn, proto.ErrInvalidRange)
		return nil
	}

	responseWriter, err := transform.NewInstantPeriodsWriter(transform.InstantPeriodsWriterOptions{
		Dst:            conn,
		GroupBy:        req.GroupBy,
		AggregateFuncs: req.AggregateFuncs,
		FirstHourOfDay: req.FirstHourOfDay,
	})
	if err != nil {
		reply(conn, proto.ErrUnexpected)
		return nil
	}

	return s.rangeScan(rangeScanReq{
		MetricID:       req.MetricID,
		MetricType:     qb.Instant,
		Since:          uint32(since.Unix()),
		Until:          uint32(until.Unix()),
		Conn:           conn,
		ResponseWriter: responseWriter,
	})
}

func (s *Database) ListCumulativePeriods(conn net.Conn, req proto.ListCumulativePeriodsReq) error {
	since, until := timeBoundsOfAggregation(req.Since, req.Until, req.GroupBy, req.FirstHourOfDay)
	if since.After(until) {
		reply(conn, proto.ErrInvalidRange)
		return nil
	}

	responseWriter, err := transform.NewCumulativePeriodsWriter(transform.CumulativePeriodsWriterOptions{
		Dst:            conn,
		GroupBy:        req.GroupBy,
		FirstHourOfDay: req.FirstHourOfDay,
	})
	if err != nil {
		reply(conn, proto.ErrUnexpected)
		return nil
	}

	return s.rangeScan(rangeScanReq{
		MetricID:       req.MetricID,
		MetricType:     qb.Cumulative,
		Since:          uint32(since.Unix()),
		Until:          uint32(until.Unix()),
		Conn:           conn,
		ResponseWriter: responseWriter,
	})
}

type rangeScanReq struct {
	MetricID       uint32
	MetricType     qb.MetricType
	Since          uint32
	Until          uint32
	Conn           io.Writer
	ResponseWriter atree.PeriodsWriter
}

func (s *Database) rangeScan(req rangeScanReq) error {
	resultCh := make(chan rangeScanResult, 1)

	s.appendJobToWorkerQueue(tryRangeScanReq{
		MetricID:       req.MetricID,
		Since:          req.Since,
		Until:          req.Until,
		MetricType:     req.MetricType,
		ResponseWriter: req.ResponseWriter,
		ResultCh:       resultCh,
	})

	result := <-resultCh

	switch result.ResultCode {
	case QueryDone:
		req.ResponseWriter.Close()

	case UntilFound:
		err := s.atree.ContinueRangeScan(atree.ContinueRangeScanReq{
			FracDigits:     result.FracDigits,
			ResponseWriter: req.ResponseWriter,
			LastPageNo:     result.LastPageNo,
			Since:          req.Since,
		})
		s.metricRUnlock(req.MetricID)

		if err != nil {
			reply(req.Conn, proto.ErrUnexpected)
		} else {
			req.ResponseWriter.Close()
		}

	case UntilNotFound:
		err := s.atree.RangeScan(atree.RangeScanReq{
			FracDigits:     result.FracDigits,
			ResponseWriter: req.ResponseWriter,
			RootPageNo:     result.RootPageNo,
			Since:          req.Since,
			Until:          req.Until,
		})
		s.metricRUnlock(req.MetricID)

		if err != nil {
			reply(req.Conn, proto.ErrUnexpected)
		} else {
			req.ResponseWriter.Close()
		}

	case NoMetric:
		reply(req.Conn, proto.ErrNoMetric)

	case WrongMetricType:
		reply(req.Conn, proto.ErrWrongMetricType)

	default:
		qb.Abort(qb.WrongResultCodeBug, ErrWrongResultCodeBug)
	}
	return nil
}

type fullScanReq struct {
	MetricID       uint32
	MetricType     qb.MetricType
	Conn           io.Writer
	ResponseWriter atree.PeriodsWriter
}

func (s *Database) fullScan(req fullScanReq) error {
	resultCh := make(chan fullScanResult, 1)

	s.appendJobToWorkerQueue(tryFullScanReq{
		MetricID:       req.MetricID,
		MetricType:     req.MetricType,
		ResponseWriter: req.ResponseWriter,
		ResultCh:       resultCh,
	})

	result := <-resultCh

	switch result.ResultCode {
	case QueryDone:
		req.ResponseWriter.Close()

	case UntilFound:
		err := s.atree.ContinueFullScan(atree.ContinueFullScanReq{
			FracDigits:     result.FracDigits,
			ResponseWriter: req.ResponseWriter,
			LastPageNo:     result.LastPageNo,
		})
		s.metricRUnlock(req.MetricID)
		if err != nil {
			reply(req.Conn, proto.ErrUnexpected)
		} else {
			req.ResponseWriter.Close()
		}

	case NoMetric:
		reply(req.Conn, proto.ErrNoMetric)

	case WrongMetricType:
		reply(req.Conn, proto.ErrWrongMetricType)

	default:
		qb.Abort(qb.WrongResultCodeBug, ErrWrongResultCodeBug)
	}
	return nil
}

func (s *Database) ListCurrentValues(conn net.Conn, req proto.ListCurrentValuesReq) error {
	responseWriter := transform.NewCurrentValueWriter(conn)
	defer responseWriter.Close()

	resultCh := make(chan struct{})

	s.appendJobToWorkerQueue(tryListCurrentValuesReq{
		MetricIDs:      req.MetricIDs,
		ResponseWriter: responseWriter,
		ResultCh:       resultCh,
	})

	<-resultCh
	return nil
}