Initial commit

FLock.go

@@ -0,0 +1,36 @@
+package bitcaspy
+
+import (
+	"fmt"
+	"os"
+
+	"golang.org/x/sys/unix"
+)
+
+func getFLock(flockfile string)(*os.File, error) {
+	flockF, err := os.Create(flockfile)
+	if err != nil {
+		return nil, fmt.Errorf("cannot create lock file %q: %w", flockF, err)
+	}
+
+	if err := unix.Flock(int(flockF.Fd()), unix.LOCK_EX|unix.LOCK_NB); err != nil {
+		return nil, fmt.Errorf("cannot acquire lock on file %q: %w", flockF, err)
+	}
+
+	return flockF, nil
+}
+
+func destroyFLock(flockF *os.File) error {
+	if err := unix.Flock(int(flockF.Fd()), unix.LOCK_UN); err != nil {
+		return fmt.Errorf("cannot unlock lock on file %q: %w", flockF.Name(), err)
+	}
+	// Close any open fd.
+	if err := flockF.Close(); err != nil {
+		return fmt.Errorf("cannot close fd on file %q: %w", flockF.Name(), err)
+	}
+	// Remove the lock file from the filesystem.
+	if err := os.Remove(flockF.Name()); err != nil {
+		return fmt.Errorf("cannot remove file %q: %w", flockF.Name(), err)
+	}
+	return nil
+}

KeyDir.go

@@ -0,0 +1,48 @@
+package bitcaspy
+
+import (
+	"encoding/gob"
+	"os"
+)
+
+type KeyDir map[string]Meta
+
+type Meta struct {
+	id int
+	value_sz int
+	value_pos int
+	tstamp int
+}
+
+func (k *KeyDir) Encode(fPath string) error {
+	file, err := os.Create(fPath)
+	if err != nil {
+		return err
+	}
+	defer file.Close()
+
+	encoder := gob.NewEncoder(file)
+
+	if err := encoder.Encode(k); err != nil {
+		return err
+	}
+
+	return	nil
+}
+
+func (k *KeyDir) Decode(fPath string) (error) {
+	file, err := os.Create(fPath)
+	if err != nil {
+		return err
+	}
+	defer file.Close()
+
+	decoder := gob.NewDecoder(file)
+
+	if err := decoder.Decode(k); err != nil {
+		return err
+	}
+
+	return	nil
+
+}

compact.go

@@ -0,0 +1,2 @@
+package bitcaspy
+

go.mod

@@ -0,0 +1,8 @@
+module rohit.com
+
+go 1.21.3
+
+require (
+	github.com/zerodha/logf v0.5.5 // indirect
+	golang.org/x/sys v0.13.0 // indirect
+)

go.sum

@@ -0,0 +1,4 @@
+github.com/zerodha/logf v0.5.5 h1:AhxHlixHNYwhFjvlgTv6uO4VBKYKxx2I6SbHoHtWLBk=
+github.com/zerodha/logf v0.5.5/go.mod h1:HWpfKsie+WFFpnUnUxelT6Z0FC6xu9+qt+oXNMPg6y8=
+golang.org/x/sys v0.13.0 h1:Af8nKPmuFypiUBjVoU9V20FiaFXOcuZI21p0ycVYYGE=
+golang.org/x/sys v0.13.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=

header.go

@@ -0,0 +1,42 @@
+package bitcaspy
+
+import (
+	"bytes"
+	"encoding/binary"
+	"hash/crc32"
+	"time"
+)
+
+type Record struct {
+	Header Header
+	Key    string
+	Value  []byte
+}
+
+type Header struct {
+	crc    uint32
+	tstamp uint32
+	expiry uint32
+	ksz    uint32
+	vsz    uint32
+}
+
+func (h *Header) Encode(buf *bytes.Buffer) error {
+	return binary.Write(buf, binary.LittleEndian, h)
+}
+
+// Decode takes a record object decodes the binary value the buffer.
+func (h *Header) Decode(record []byte) error {
+	return binary.Read(bytes.NewReader(record), binary.LittleEndian, h)
+}
+
+func (r *Record) isExpired() bool {
+	if r.Header.expiry == 0 {
+		return false
+	}
+	return int64(r.Header.expiry) < time.Now().Unix()  
+}
+
+func (r *Record) isValidChecksum() bool {
+	return crc32.ChecksumIEEE(r.Value) == r.Header.crc
+}

internal/datafile.go

@@ -0,0 +1,111 @@
+package datafile
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+	"sync"
+)
+
+const (
+	ACTIVE_DATAFILE = "bitcaspy%d.db"
+)
+
+type DataFile struct {
+	sync.RWMutex
+
+	writer *os.File
+	reader *os.File
+	id     int
+
+	offset int
+}
+
+// New initializes a new DataFile for storing the data in the database
+// Only one datafile ca be active at a time
+func New(dir string, index int) (*DataFile, error) {
+	// If the new file is not already present because of filled file of previous operation otherwise use the present active file
+
+	path := filepath.Join(dir, fmt.Sprintf(ACTIVE_DATAFILE, index))
+	writer, err := os.OpenFile(path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
+
+	if err != nil {
+		return nil, fmt.Errorf("error opening file for writing db: %w", err)
+	}
+
+	reader, err := os.Open(path)
+	if err != nil {
+		return nil, fmt.Errorf("error opening file for reading db: %w", err)
+	}
+
+	stat, err := writer.Stat()
+	if err != nil {
+		return nil, fmt.Errorf("error fetching file stats: %v", err)
+	}
+
+	df := &DataFile{
+		writer: writer,
+		reader: reader,
+		id:     index,
+		offset: int(stat.Size()),
+	}
+	return df, nil
+}
+
+func (d *DataFile) ID() int {
+	return d.id
+}
+
+func (d *DataFile) Size() (int64, error) {
+	stat, err := d.writer.Stat()
+	if err != nil {
+		return -1, fmt.Errorf("error getting file size: %v", err)
+	}
+
+	return stat.Size(), nil
+}
+
+func (d *DataFile) Read(pos int, size int) ([]byte, error) {
+	start := int64(pos - size)
+
+	record := make([]byte, size)
+
+	n, err := d.reader.ReadAt(record, start)
+	if err != nil {
+		return nil, err
+	}
+
+	if n != size {
+		return nil, fmt.Errorf("error fetching record, invalid size")
+	}
+
+	return record, nil
+}
+
+func (d *DataFile) Write(data []byte) (int, error) {
+	if _, err := d.writer.Write(data); err != nil {
+		return 0, err
+	}
+
+	offset := d.offset
+
+	offset += len(data)
+
+	return offset, nil
+}
+
+func (d *DataFile) Close() error {
+	if err := d.writer.Close(); err != nil {
+		return err
+	}
+
+	if err := d.reader.Close(); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (d *DataFile) Sync () error {
+	return d.writer.Sync()
+}

main.go

@@ -0,0 +1,46 @@
+package bitcaspy
+
+import (
+	"bytes"
+	"fmt"
+	"os"
+	"path/filepath"
+	"sync"
+	"time"
+
+	"github.com/zerodha/logf"
+	datafile "rohit.com/internal"
+)
+
+const (
+	LOCKFILE   = "bitcaspy.lock"
+	HINTS_FILE = "bitcaspy.hints"
+)
+
+type BitCaspy struct{
+	sync.RWMutex
+
+	lo logf.Logger
+	bufPool sync.Pool
+
+	KeyDir KeyDir
+	df *datafile.DataFile
+	stale map[int]*datafile.DataFile
+	flockF *os.File 
+}
+
+func initLogger(debug bool) logf.Logger {
+	opts := logf.Opts{EnableCaller: true}
+	if debug {
+		opts.Level = logf.DebugLevel
+	}
+	return logf.New(opts)
+}
+
+func Init()(*BitCaspy, error){
+	var (
+		flockF *os.File
+		stale = map[int]*datafile.DataFile{}
+
+	)
+}

merge.go

@@ -0,0 +1 @@
+package bitcaspy