qb/database/database.go

package database

import (
	"errors"
	"fmt"
	"io"
	"log"
	"net"
	"os"
	"path/filepath"
	"sync"
	"time"

	bin "gordenko.dev/dima/bin/little"
	"gordenko.dev/dima/qb"
	"gordenko.dev/dima/qb/atree"
	"gordenko.dev/dima/qb/freelist"
	"gordenko.dev/dima/qb/txlog"
)

func JoinSnapshotFileName(dir string, logNumber int) string {
	return filepath.Join(dir, fmt.Sprintf("%d.snapshot", logNumber))
}

// type metricLockEntry struct {
// 	XLock     bool
// 	RLocks    int
// 	WaitQueue []any
// }

type Database struct {
	mutex           sync.Mutex
	workerSignalCh  chan struct{}
	workerQueue     []any
	rLocksToRelease []uint32
	metrics         map[uint32]*_metric
	//metricLockEntries map[uint32]*metricLockEntry
	dataFreeList   *freelist.FreeList
	indexFreeList  *freelist.FreeList
	dir            string
	databaseName   string
	snapshotNumber int
	txlog          *txlog.Writer
	atree          *atree.Atree
	tcpPort        int
	logfile        *os.File
	logger         *log.Logger
	exitCh         chan struct{}
	waitGroup      *sync.WaitGroup
}

type Options struct {
	TCPPort      int
	Dir          string
	DatabaseName string
	RedoDir      string
	Logfile      *os.File
	ExitCh       chan struct{}
	WaitGroup    *sync.WaitGroup
}

func New(opt Options) (_ *Database, err error) {
	if opt.TCPPort <= 0 {
		return nil, errors.New("TCPPort option is required")
	}
	if opt.Dir == "" {
		return nil, errors.New("Dir option is required")
	}
	if opt.DatabaseName == "" {
		return nil, errors.New("DatabaseName option is required")
	}
	if opt.RedoDir == "" {
		return nil, errors.New("RedoDir option is required")
	}
	if opt.Logfile == nil {
		return nil, errors.New("Logfile option is required")
	}
	if opt.ExitCh == nil {
		return nil, errors.New("ExitCh option is required")
	}
	if opt.WaitGroup == nil {
		return nil, errors.New("WaitGroup option is required")
	}

	// deltaFreeListFile, err := os.OpenFile(
	// 	filepath.Join(opt.Dir, opt.DatabaseName+".deltafree"),
	// 	os.O_RDWR|os.O_CREATE,
	// 	0666,
	// )
	// if err != nil {
	// 	return nil, err
	// }

	dataFreeList, err := freelist.New(freelist.Options{
		PageSize:      2048,
		BaseFilePath:  filepath.Join(opt.Dir, opt.DatabaseName+".free_data_base"),
		DeltaFilePath: filepath.Join(opt.Dir, opt.DatabaseName+".free_data_delta"),
	})

	indexFreeList, err := freelist.New(freelist.Options{
		PageSize:      2048,
		BaseFilePath:  filepath.Join(opt.Dir, opt.DatabaseName+".free_index_base"),
		DeltaFilePath: filepath.Join(opt.Dir, opt.DatabaseName+".free_index_delta"),
	})

	s := &Database{
		workerSignalCh: make(chan struct{}, 1),
		dir:            opt.Dir,
		databaseName:   opt.DatabaseName,
		metrics:        make(map[uint32]*_metric),
		//metricLockEntries: make(map[uint32]*metricLockEntry),
		dataFreeList:  dataFreeList,
		indexFreeList: indexFreeList,
		tcpPort:       opt.TCPPort,
		logfile:       opt.Logfile,
		logger:        log.New(opt.Logfile, "", log.LstdFlags),
		exitCh:        opt.ExitCh,
		waitGroup:     opt.WaitGroup,
	}
	return s, nil
}

func (s *Database) ListenAndServe() (err error) {
	listener, err := net.Listen("tcp", fmt.Sprintf(":%d", s.tcpPort))
	if err != nil {
		return fmt.Errorf("net.Listen: %s; port=%d", err, s.tcpPort)
	}

	s.atree, err = atree.New(atree.Options{
		Dir:          s.dir,
		DatabaseName: s.databaseName,
	})
	if err != nil {
		return fmt.Errorf("atree.New: %s", err)
	}
	s.atree.Run()

	go s.worker()

	s.recovery()

	s.logger.Println("database started")
	for {
		// Listen for an incoming connection.
		conn, err := listener.Accept()
		if err != nil {
			s.logger.Printf("listener.Accept: %s\n", err)
			time.Sleep(time.Second)
		} else {
			go s.handleTCPConn(conn)
		}
	}
}

func (s *Database) 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.txlog, err = txlog.NewWriter(txlog.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.txlog.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)
	// 		}
	// 	}
	// }
}

//

// зробити object?

func (s *Database) replayChanges(fileName string) error {
	walReader, err := txlog.NewReader(txlog.ReaderOptions{
		FileName:   fileName,
		BufferSize: 1024 * 1024,
	})
	if err != nil {
		return err
	}

	var metrics map[uint32]*ReplayMetric

	for {
		records, isLast, err := walReader.NextPacket()
		if err != nil {
			return err
		}

		if isLast {
			if len(records) > 0 {
				var appendRecords []txlog.WALRecordAppendMeasures
				for _, untyped := range records {
					rec, ok := untyped.(txlog.WALRecordAppendMeasures)
					if ok {
						appendRecords = append(appendRecords, rec)
					}
				}
				// prepare pages and rewrite
				indexPages, dataPages := pagesToWriteFromWALRecords(metrics, appendRecords)

				err = s.txlog.WritePagesToAtree(indexPages, dataPages)
				if err != nil {
					return err
				}
			}
			// для останнього rec - повторюємо запис сторінок
			// rewrite write pages
			//s.txlog.
			//return nil
		}

		for _, record := range records {
			if err = s.replayWALRecord(record); err != nil {
				return err
			}
		}
	}
}

func (s *Database) replayWALRecord(untyped any) error {
	switch rec := untyped.(type) {
	case txlog.AddedMetric:
		s.addMetric(rec)

	case txlog.DeletedMetric:
		// pages add and delete to tmp file

		// delete(s.metrics, rec.MetricID)
		// if len(rec.FreePageNumbers) > 0 {
		// 	s.freeList.AddPages(rec.FreePageNumbers)
		// }

	// case txlog.AppendedMeasure:
	// 	metric, ok := s.metrics[rec.MetricID]
	// 	if ok {
	// 		metric.Timestamps.Append(rec.Timestamp)
	// 		metric.Values.Append(rec.Value)

	// 		if metric.Since == 0 {
	// 			metric.Since = rec.Timestamp
	// 			metric.SinceValue = rec.Value
	// 		}

	// 		metric.Until = rec.Timestamp
	// 		metric.UntilValue = rec.Value
	// 	}

	case txlog.WALRecordAppendMeasures:
		metric, ok := s.metrics[rec.MetricID]
		if !ok {
			qb.Abort(qb.MetricNotFoundDuringReplay, nil)
		}
		//metric.ReplayAppendedMeasures(rec) FIX

	// case txlog.DeletedMeasures:
	// 	metric, ok := s.metrics[rec.MetricID]
	// 	if ok {
	// 		metric.DeleteMeasures()
	// 		if len(rec.FreePageNumbers) > 0 {
	// 			s.freeList.AddPages(rec.FreePageNumbers)
	// 		}
	// 	}

	default:
		qb.Abort(qb.UnknownTxLogRecordTypeBug,
			fmt.Errorf("bug: unknown record type %T in TransactionLog", rec))
	}
	return nil
}

type ReplayMetric struct {
	metricType      qb.MetricType
	fracDigits      byte
	lastPageNo      uint32
	buf             []byte
	databuf         []byte // якщо розмір buf == DataPageSize, то databuf буде менше (мінус футер)
	tSize           int
	vSize           int
	indexLevelTails []txlog.IndexLevelTail // root - last element
}

func (s *ReplayMetric) ReadFrom(r io.Reader) (err error) {
	metricType, err := bin.ReadByte(r)
	if err != nil {
		return
	}
	s.metricType = qb.MetricType(metricType)
	s.fracDigits, err = bin.ReadByte(r)
	if err != nil {
		return
	}
	s.lastPageNo, err = bin.ReadUint32(r)
	if err != nil {
		return
	}
	s.tSize, err = bin.ReadUint16AsInt(r)
	if err != nil {
		return
	}
	s.vSize, err = bin.ReadUint16AsInt(r)
	if err != nil {
		return
	}
	s.buf, s.databuf = allocateBuffers(s.tSize + s.vSize)
	//
	err = bin.ReadNInto(r, s.databuf[len(s.databuf)-s.tSize:])
	if err != nil {
		return
	}
	err = bin.ReadNInto(r, s.databuf[:s.vSize])
	if err != nil {
		return
	}
	//
	levelsCount, err := bin.ReadVarSize(r)
	if err != nil {
		return
	}
	for range levelsCount {
		var (
			buf   = make([]byte, atree.IndexPageSize)
			count int
		)
		count, err = bin.ReadVarSize(r)
		if err != nil {
			return
		}
		err = bin.ReadNInto(r, buf[:count*indexRecordSize])
		if err != nil {
			return
		}
		s.indexLevelTails = append(s.indexLevelTails, txlog.IndexLevelTail{
			Buffer:       buf,
			RecordsCount: count,
		})
	}
	return
}

func (s *ReplayMetric) ReplayAppendedMeasures(rec txlog.WALRecordAppendMeasures) {
	// Додати в index level tails недостаючі дані, або замінити
	for levelIdx, change := range rec.ChangedIndexLevels {
		if levelIdx < len(s.indexLevelTails) {
			level := s.indexLevelTails[levelIdx]
			if change.CompletedIndexPage != nil {
				// попередній хвіст індексного рівня перетворився на сторінку,
				// отже TailRecords - це новий хвіст
				// або копіюю дані, або алокую більший буфер?
				copy(level.Buffer, change.TailRecords)
				level.RecordsCount = len(change.TailRecords) / indexRecordSize
			} else {
				// TailRecords - це нові дані, які треба додати
				pos := level.RecordsCount * indexRecordSize
				copy(level.Buffer[pos:], change.TailRecords)
			}
			s.indexLevelTails[levelIdx] = level
		} else {
			var (
				recordsCount = len(change.TailRecords) / indexRecordSize
				buffer       = make([]byte, atree.IndexPageSize)
			)
			copy(buffer, change.TailRecords)
			s.indexLevelTails = append(s.indexLevelTails, txlog.IndexLevelTail{
				Buffer:       buffer,
				RecordsCount: recordsCount,
			})
		}
	}

	// Перезаписати в buffer timestamps та values. Якщо розмір буферу недостатній -
	// спочатку створити більший

	var (
		// CompletedDataPage є полюбому, оскільки це Record із мінімум одною заповненою
		// data сторінкою. Отже TailTimestamps і TailValues - це нові хвости data рівня.

		requiredSpace = len(rec.TailTimestamps) + len(rec.TailValues)
		databuf       []byte
	)
	if requiredSpace < len(s.buf) {
		databuf = s.buf
		// fix - update sizes?
	} else {
		var buf []byte
		buf, databuf = allocateBuffers(requiredSpace)
		copy(databuf, rec.TailValues)
		pos := len(databuf) - len(rec.TailTimestamps)
		copy(databuf[pos:], rec.TailTimestamps)
		// fix - replace and remeber sizes?

		s.buf = buf
	}

	//rec.TailTimestamps
	//rec.TailValues
}

func (s *Database) rewritePagesFromLastWALPacket(list []any) error {
	// var (
	// 	indexPages []txlog.PageToWrite
	// 	dataPages  []txlog.PageToWrite
	// )
	// for _, untyped := range list {
	// 	switch rec := untyped.(type) {
	// 	case txlog.WALRecordAppendMeasures:
	// 		metric, ok := s.metrics[rec.MetricID]
	// 		if !ok {
	// 			qb.Abort(qb.MetricNotFoundDuringReplay, nil)
	// 		}
	// 		//metric.ReplayAppendedMeasures(rec)

	// 	default:
	// 		qb.Abort(qb.UnknownTxLogRecordTypeBug,
	// 			fmt.Errorf("bug: unknown record type %T in TransactionLog", rec))
	// 	}
	// }
	return nil
}

func pagesToWriteFromWALRecords(metrics map[uint32]*ReplayMetric, records []txlog.WALRecordAppendMeasures) (indexPages []txlog.PageToWrite, dataPages []txlog.PageToWrite) {
	for _, rec := range records {
		metric, ok := metrics[rec.MetricID]
		if !ok {
			qb.Abort(qb.MetricNotFoundDuringReplay, nil)
		}
		p := rec.CompletedDataPage
		// буфер має бути розміру сторінки
		if len(metric.buf) != atree.DataPageSize {
			metric.buf, metric.databuf = growBuffers(growBuffersIn{
				databuf:       metric.databuf,
				tSize:         metric.tSize,
				vSize:         metric.vSize,
				requiredSpace: atree.DataPageSize,
			})
		}
		// дописую дані по зміщенню
		pos := len(metric.databuf) - p.TimestampsOffset - len(p.Timestamps)
		copy(metric.databuf[pos:], p.Timestamps)
		copy(metric.databuf[p.ValuesOffset:], p.Values)
		//
		metric.tSize = p.TimestampsOffset + len(p.Timestamps)
		metric.tSize = p.ValuesOffset + len(p.Values)

		// fix - заполнить страницу
		// fix - check checksum

		dataPages = append(dataPages, txlog.PageToWrite{
			PageNo:  p.PageNo,
			Content: metric.buf,
		})

		for _, x := range rec.DataPages {
			dataPages = append(dataPages, txlog.PageToWrite{
				PageNo:  x.PageNo,
				Content: x.Content,
			})
		}

		for levelIdx, changedLevel := range rec.ChangedIndexLevels {
			p := changedLevel.CompletedIndexPage
			if p != nil {
				level := metric.indexLevelTails[levelIdx]
				copy(level.Buffer[level.RecordsCount*indexRecordSize:], p.Records)

				// fix - заполнить страницу
				// fix - check checksum

				indexPages = append(dataPages, txlog.PageToWrite{
					PageNo:  p.PageNo,
					Content: level.Buffer,
				})
			}

			for _, x := range changedLevel.IndexPages {
				indexPages = append(dataPages, txlog.PageToWrite{
					PageNo:  x.PageNo,
					Content: x.Content,
				})
			}
		}
	}
	return
}

// FIX
// УВАГА!
// Якщо буфер metric.buffer < розміру сторінки - для timestamps доступний весь буфер.
// Якщо буфер досяг розміру сторінки - в timestamps я передаю buffer[:DataPagePayloadSize]

var dataBufferSizes = []int{
	1024,
	2048,
	4096,
	8192,
}

func calculateDataBufferSize(payloadSize int) int {
	for _, bufferSize := range dataBufferSizes {
		if payloadSize <= bufferSize {
			return bufferSize
		}
	}
	return atree.DataPageSize
}

func allocateBuffers(requiredSpace int) (buf, databuf []byte) {
	bufferSize := calculateDataBufferSize(requiredSpace)
	buf = make([]byte, bufferSize)
	if bufferSize == atree.DataPageSize {
		databuf = buf[:atree.DataPagePayloadSize]
	} else {
		databuf = buf
	}
	return
}

// src - databuf
type growBuffersIn struct {
	databuf       []byte
	tSize         int
	vSize         int
	requiredSpace int
}

func growBuffers(in growBuffersIn) (buf, databuf []byte) {
	buf, databuf = allocateBuffers(in.requiredSpace)
	copy(databuf[len(databuf)-in.tSize:], in.databuf[len(in.databuf)-in.tSize:])
	copy(databuf, in.databuf[:in.vSize])
	return
}

//func (s *Database) verifySnapshot(fileName string) (_ bool, err error) {
// 	file, err := os.Open(fileName)
// 	if err != nil {
// 		return
// 	}
// 	defer file.Close()

// 	stat, err := file.Stat()
// 	if err != nil {
// 		return
// 	}

// 	if stat.Size() <= 4 {
// 		return false, nil
// 	}

// 	var (
// 		payloadSize = stat.Size() - 4
// 		hash        = crc32.NewIEEE()
// 	)

// 	_, err = io.CopyN(hash, file, payloadSize)
// 	if err != nil {
// 		return
// 	}
// 	calculatedCRC := hash.Sum32()

// 	storedCRC, err := bin.ReadUint32(file)
// 	if err != nil {
// 		return
// 	}
// 	if storedCRC != calculatedCRC {
// 		return false, fmt.Errorf("strored CRC %d not equal calculated CRC %d",
// 			storedCRC, calculatedCRC)
// 	}
// 	return true, nil
// }