package store

import (
	"encoding/binary"
	"errors"
	"fmt"
	"hash/crc32"
	"io"
	"os"
	"path/filepath"
	"sort"
	"strings"
)

// The WAL is a sequence of segments named seg-NNN.dat. Each record is
//
//	uint32 length | uint32 crc32(payload) | payload
//
// A serial (segment<<32 | offset) is the physical address of the record's
// header, so reading an event is one open segment, one ReadAt for the header
// and one ReadAt for the payload. The log is never rewritten.
const (
	walHeaderSize = 8
	// MaxSegSize bounds a segment so its offset fits in the 32 low bits of a
	// serial. 256 segments of 1 GiB is the ceiling of the uint40 address.
	MaxSegSize = 1 << 30
	// maxRecord bounds a single record so a corrupt length cannot ask for an
	// absurd allocation.
	maxRecord = 64 << 20
	syncEvery = 100
)

var (
	ErrCorruptWAL = errors.New("store: corrupt WAL record")
	ErrShortRead  = errors.New("store: short WAL read")
)

type wal struct {
	dir    string
	segs   []*os.File
	cur    int
	off    int64
	wrote  int
	noSync bool
}

func openWAL(dir string) (*wal, error) {
	if err := os.MkdirAll(dir, 0o755); err != nil {
		return nil, err
	}
	entries, err := os.ReadDir(dir)
	if err != nil {
		return nil, err
	}
	var names []string
	for _, e := range entries {
		n := e.Name()
		if strings.HasPrefix(n, "seg-") && strings.HasSuffix(n, ".dat") {
			names = append(names, n)
		}
	}
	sort.Strings(names)
	w := &wal{dir: dir}
	for _, name := range names {
		f, err := os.OpenFile(filepath.Join(dir, name), os.O_RDWR, 0o644)
		if err != nil {
			w.Close()
			return nil, err
		}
		w.segs = append(w.segs, f)
	}
	if len(w.segs) == 0 {
		if err := w.newSeg(); err != nil {
			return nil, err
		}
	} else {
		w.cur = len(w.segs) - 1
		info, err := w.segs[w.cur].Stat()
		if err != nil {
			return nil, err
		}
		if err := w.repairTail(info.Size()); err != nil {
			return nil, err
		}
	}
	return w, nil
}

// repairTail finds the end of the last complete record in the current segment
// and truncates a torn tail left by a crash.
func (w *wal) repairTail(size int64) error {
	f := w.segs[w.cur]
	hdr := make([]byte, walHeaderSize)
	payload := make([]byte, 0, 64<<10)
	off := int64(0)
	for off < size {
		if size-off < walHeaderSize {
			break
		}
		if _, err := f.ReadAt(hdr, off); err != nil {
			break
		}
		length := int64(binary.BigEndian.Uint32(hdr[0:4]))
		if length <= 0 || length > maxRecord || off+walHeaderSize+length > size {
			break
		}
		if int64(cap(payload)) < length {
			payload = make([]byte, length)
		}
		payload = payload[:length]
		if _, err := f.ReadAt(payload, off+walHeaderSize); err != nil {
			break
		}
		if crc32.ChecksumIEEE(payload) != binary.BigEndian.Uint32(hdr[4:8]) {
			break
		}
		off += walHeaderSize + length
	}
	if off != size {
		if err := f.Truncate(off); err != nil {
			return err
		}
		if _, err := f.Seek(off, io.SeekStart); err != nil {
			return err
		}
	}
	w.off = off
	return nil
}

func (w *wal) newSeg() error {
	n := len(w.segs)
	f, err := os.OpenFile(filepath.Join(w.dir, fmt.Sprintf("seg-%03d.dat", n)), os.O_RDWR|os.O_CREATE, 0o644)
	if err != nil {
		return err
	}
	w.segs = append(w.segs, f)
	w.cur = n
	w.off = 0
	return nil
}

// Pos returns the serial the next record will be written at.
func (w *wal) Pos() uint64 { return uint64(w.cur)<<32 | uint64(w.off) }

// Append writes a record and returns its serial.
func (w *wal) Append(data []byte) (uint64, error) {
	need := int64(walHeaderSize + len(data))
	if w.off+need > MaxSegSize {
		if err := w.Sync(); err != nil {
			return 0, err
		}
		if err := w.newSeg(); err != nil {
			return 0, err
		}
	}
	ser := uint64(w.cur)<<32 | uint64(w.off)
	hdr := make([]byte, walHeaderSize)
	binary.BigEndian.PutUint32(hdr[0:4], uint32(len(data)))
	binary.BigEndian.PutUint32(hdr[4:8], crc32.ChecksumIEEE(data))
	f := w.segs[w.cur]
	if _, err := f.WriteAt(hdr, w.off); err != nil {
		return 0, err
	}
	if _, err := f.WriteAt(data, w.off+walHeaderSize); err != nil {
		return 0, err
	}
	w.off += need
	w.wrote++
	if !w.noSync && w.wrote >= syncEvery {
		w.wrote = 0
		if err := f.Sync(); err != nil {
			return 0, err
		}
	}
	return ser, nil
}

// Read returns the payload stored at ser.
func (w *wal) Read(ser uint64) ([]byte, error) {
	seg := int(ser >> 32)
	off := int64(ser & 0xffffffff)
	if seg < 0 || seg >= len(w.segs) {
		return nil, ErrShortRead
	}
	f := w.segs[seg]
	hdr := make([]byte, walHeaderSize)
	if _, err := f.ReadAt(hdr, off); err != nil {
		return nil, err
	}
	length := int64(binary.BigEndian.Uint32(hdr[0:4]))
	if length <= 0 || length > maxRecord {
		return nil, ErrCorruptWAL
	}
	data := make([]byte, length)
	if _, err := f.ReadAt(data, off+walHeaderSize); err != nil {
		return nil, err
	}
	if crc32.ChecksumIEEE(data) != binary.BigEndian.Uint32(hdr[4:8]) {
		return nil, ErrCorruptWAL
	}
	return data, nil
}

// ForEachFrom calls fn for every record at or after pos. fn returning false
// stops the scan.
func (w *wal) ForEachFrom(pos uint64, fn func(ser uint64, data []byte) bool) error {
	seg := int(pos >> 32)
	off := int64(pos & 0xffffffff)
	if seg >= len(w.segs) {
		return nil
	}
	hdr := make([]byte, walHeaderSize)
	var payload []byte
	for ; seg < len(w.segs); seg++ {
		f := w.segs[seg]
		if seg == int(pos>>32) {
			off = int64(pos & 0xffffffff)
		} else {
			off = 0
		}
		info, err := f.Stat()
		if err != nil {
			return err
		}
		size := info.Size()
		for off < size {
			if size-off < walHeaderSize {
				break
			}
			if _, err := f.ReadAt(hdr, off); err != nil {
				return err
			}
			length := int64(binary.BigEndian.Uint32(hdr[0:4]))
			if length <= 0 || length > maxRecord || off+walHeaderSize+length > size {
				break
			}
			if int64(cap(payload)) < length {
				payload = make([]byte, length)
			}
			payload = payload[:length]
			if _, err := f.ReadAt(payload, off+walHeaderSize); err != nil {
				return err
			}
			if crc32.ChecksumIEEE(payload) != binary.BigEndian.Uint32(hdr[4:8]) {
				break
			}
			ser := uint64(seg)<<32 | uint64(off)
			if !fn(ser, payload) {
				return nil
			}
			off += walHeaderSize + length
		}
	}
	return nil
}

func (w *wal) Sync() error {
	if len(w.segs) == 0 {
		return nil
	}
	return w.segs[w.cur].Sync()
}

func (w *wal) Close() error {
	var first error
	for _, f := range w.segs {
		if err := f.Sync(); err != nil && first == nil {
			first = err
		}
		if err := f.Close(); err != nil && first == nil {
			first = err
		}
	}
	w.segs = nil
	return first
}