package store

import (
	"bytes"
	"errors"
	"fmt"
	"math"
	"os"
	"path/filepath"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"

	"git.smesh.lol/torly/pkg/nostr"
)

var (
	// ErrDuplicate means the event id is already stored.
	ErrDuplicate = errors.New("store: duplicate event")
	// ErrReplaced means a newer replaceable event already holds the slot.
	ErrReplaced = errors.New("store: replaced by a newer event")
	// ErrDeleted means the event id carries a NIP-09 tombstone.
	ErrDeleted = errors.New("store: event was deleted")
	// ErrNotFound means no event matches.
	ErrNotFound = errors.New("store: event not found")
	// ErrClosed means the engine has been closed.
	ErrClosed = errors.New("store: closed")
)

// DefaultMaxResults bounds an unbounded query so that a single REQ cannot ask
// the relay to decode an entire database.
const DefaultMaxResults = 5000

// Options configures an Engine.
type Options struct {
	// MaxResults is the result cap when a filter carries no limit (or a larger
	// one). Zero selects DefaultMaxResults.
	MaxResults int

	// BulkImport defers the social graph indexes until RebuildGraph is called.
	// A bulk importer writes millions of events; maintaining and backfilling
	// edges after every one of them is far slower than building the graph in a
	// single pass once all targets are present.
	BulkImport bool
}

// Engine is a single-writer storage engine. All public methods serialize on
// one mutex: the relay is expected to verify signatures outside it and to let
// this engine own the ordering of writes.
type Engine struct {
	dir string
	mu  sync.Mutex

	w    *wal
	ckpt *checkpoint

	nextPubSer uint64
	lastSer    uint64
	saveN      int
	maxResults int
	bulk       bool
	bulkRep    map[string]uint64
	bulkIDs    map[[32]byte]uint64
	bulkPub    map[string]uint64
	removedSer map[uint64]struct{}
	closed     bool

	eid *sortedFile
	sei *sortedFile
	fpc *sortedFile
	ca  *sortedFile
	exp *sortedFile
	kc  *sortedFile
	pc  *sortedFile
	kpc *sortedFile
	tc  *sortedFile
	tkc *sortedFile
	tpc *sortedFile
	tkp *sortedFile
	wrd *sortedFile
	rep *sortedFile
	del *sortedFile

	// Pubkey serials and social graph indexes.
	pks *sortedFile
	spk *sortedFile
	epg *sortedFile
	peg *sortedFile
	eeg *sortedFile
	gee *sortedFile
	ppg *sortedFile
	gpp *sortedFile
}

// Open opens or creates a store in dir.
func Open(dir string, opts Options) (*Engine, error) {
	if opts.MaxResults <= 0 {
		opts.MaxResults = DefaultMaxResults
	}
	if err := os.MkdirAll(filepath.Join(dir, "idx"), 0o755); err != nil {
		return nil, err
	}
	w, err := openWAL(filepath.Join(dir, "wal"))
	if err != nil {
		return nil, err
	}
	w.noSync = opts.BulkImport
	e := &Engine{dir: dir, w: w, nextPubSer: 1, maxResults: opts.MaxResults, bulk: opts.BulkImport}
	if e.bulk {
		e.bulkRep = make(map[string]uint64)
		e.bulkIDs = make(map[[32]byte]uint64)
		e.removedSer = make(map[uint64]struct{})
	}
	idx := filepath.Join(dir, "idx")

	type spec struct {
		dst    **sortedFile
		name   string
		recLen int
		cmpLen int
		serOff int
	}
	last := func(n int) int { return n - SerialLen }
	specs := []spec{
		{&e.eid, "eid", eidRecLen, eidCmpLen, 3 + HashLen},
		{&e.sei, "sei", seiRecLen + seiValLen, seiCmpLen, 3},
		{&e.fpc, "fpc", fpcRecLen, fpcCmpLen, 3},
		{&e.ca, "ca", caRecLen, caCmpLen, last(caRecLen)},
		{&e.exp, "exp", expRecLen, expCmpLen, last(expRecLen)},
		{&e.kc, "kc", kcRecLen, kcCmpLen, last(kcRecLen)},
		{&e.pc, "pc", pcRecLen, pcCmpLen, last(pcRecLen)},
		{&e.kpc, "kpc", kpcRecLen, kpcCmpLen, last(kpcRecLen)},
		{&e.tc, "tc", tcRecLen, tcCmpLen, last(tcRecLen)},
		{&e.tkc, "tkc", tkcRecLen, tkcCmpLen, last(tkcRecLen)},
		{&e.tpc, "tpc", tpcRecLen, tpcCmpLen, last(tpcRecLen)},
		{&e.tkp, "tkp", tkpRecLen, tkpCmpLen, last(tkpRecLen)},
		{&e.wrd, "wrd", wrdRecLen, wrdCmpLen, last(wrdRecLen)},
		{&e.rep, "rep", repRecLen, repCmpLen, last(repRecLen)},
		{&e.del, "del", delRecLen, delCmpLen, -1},
		{&e.pks, "pks", pksRecLen, pksCmpLen, -1},
		{&e.spk, "spk", spkRecLen + spkValLen, spkCmpLen, -1},
		{&e.epg, "epg", epgRecLen, epgRecLen, -1},
		{&e.peg, "peg", pegRecLen, pegRecLen, -1},
		{&e.eeg, "eeg", eegRecLen, eegRecLen, -1},
		{&e.gee, "gee", geeRecLen, geeRecLen, -1},
		{&e.ppg, "ppg", ppgRecLen, ppgRecLen, -1},
		{&e.gpp, "gpp", gppRecLen, gppRecLen, -1},
	}
	for _, s := range specs {
		f, err := openSorted(filepath.Join(idx, s.name+".dat"), s.recLen, s.cmpLen)
		if err != nil {
			e.closeFiles()
			return nil, fmt.Errorf("store: open %s index: %w", s.name, err)
		}
		f.bulk = opts.BulkImport
		f.serOff = s.serOff
		if opts.BulkImport {
			f.removed = e.serialRemoved
		}
		*s.dst = f
	}

	c, err := openCheckpoint(filepath.Join(dir, "checkpoint.dat"))
	if err != nil {
		e.closeFiles()
		return nil, err
	}
	e.ckpt = c
	pos, nextPub := c.Get()
	if nextPub > 0 {
		e.nextPubSer = nextPub
	}

	// A checkpoint beyond the log's write position is stale: the log was
	// truncated or replaced. Rebuild from the beginning.
	if pos > w.Pos() {
		pos = 0
	}
	// Older stores without graph indexes can be upgraded explicitly with
	// `torly reindex-graph`. Do not infer that from empty pks/spk here: those
	// files are also empty while a graph rebuild is in progress, and treating
	// that as an old store would replay the whole WAL on every resume.
	if pos == 0 {
		if err := e.rebuild(); err != nil {
			e.closeFiles()
			return nil, err
		}
	} else {
		if err := e.replay(pos); err != nil {
			e.closeFiles()
			return nil, err
		}
	}
	// Bulk replay records removals as a serial set that scans consult. The
	// records are still physically present until a merge, and the set is not
	// persisted, so materialize the merge before advancing the checkpoint.
	if e.bulk && len(e.removedSer) > 0 {
		if err := e.Flush(); err != nil {
			e.closeFiles()
			return nil, err
		}
	}
	if err := e.checkpointNow(); err != nil {
		e.closeFiles()
		return nil, err
	}
	if e.bulk {
		if err := e.loadBulkIDs(); err != nil {
			e.closeFiles()
			return nil, err
		}
		if err := e.loadBulkRep(); err != nil {
			e.closeFiles()
			return nil, err
		}
	}
	return e, nil
}

// Dir returns the store directory.
func (e *Engine) Dir() string { return e.dir }

// MaxResults returns the configured result cap.
func (e *Engine) MaxResults() int { return e.maxResults }

// MaxSerial returns the WAL serial of the newest stored event.
func (e *Engine) MaxSerial() uint64 {
	e.mu.Lock()
	defer e.mu.Unlock()
	return e.lastSer
}

func (e *Engine) indexes() []*sortedFile {
	return []*sortedFile{
		e.eid, e.sei, e.fpc, e.ca, e.exp, e.kc, e.pc, e.kpc,
		e.tc, e.tkc, e.tpc, e.tkp, e.wrd, e.rep, e.del,
		e.pks, e.spk, e.epg, e.peg, e.eeg, e.gee, e.ppg, e.gpp,
	}
}

func (e *Engine) closeFiles() {
	for _, f := range e.indexes() {
		if f != nil {
			f.f.Close()
		}
	}
	if e.w != nil {
		e.w.Close()
	}
	if e.ckpt != nil {
		e.ckpt.Close()
	}
}

func (e *Engine) rebuild() error {
	for _, f := range e.indexes() {
		if err := f.Clear(); err != nil {
			return err
		}
	}
	e.nextPubSer = 1
	if e.bulk {
		e.bulkIDs = make(map[[32]byte]uint64)
		e.bulkRep = make(map[string]uint64)
		e.removedSer = make(map[uint64]struct{})
	}
	return e.replay(0)
}

// replay re-applies WAL records at or after pos. During a rebuild pos is 0
// and the indexes are empty.
func (e *Engine) replay(pos uint64) error {
	var replayErr error
	err := e.w.ForEachFrom(pos, func(ser uint64, data []byte) bool {
		ev, derr := decodeEvent(data)
		if derr != nil {
			// A record that cannot be decoded is skipped; the log is the
			// authority for bytes, not the index.
			return true
		}
		e.lastSer = ser
		if e.isTombstoned(ev.ID[:]) {
			return true
		}
		keep, kerr := e.applyReplacement(ev)
		if kerr != nil {
			replayErr = kerr
			return false
		}
		if !keep {
			return true
		}
		if ierr := e.indexEvent(ev, ser); ierr != nil {
			replayErr = ierr
			return false
		}
		if e.bulk {
			e.bulkIDs[ev.ID] = ser
			if ev.IsReplaceable() || ev.IsAddressable() {
				e.bulkRep[bulkSlotKey(ev)] = ser
			}
		}
		if ev.IsDeletion() {
			if derr := e.applyDeletion(ev); derr != nil {
				replayErr = derr
				return false
			}
		}
		return true
	})
	if err != nil {
		return err
	}
	return replayErr
}

// SaveEvent validates nothing about the event beyond its structural identity:
// the relay verifies signatures before calling this. It returns ErrDuplicate
// when the id is already stored and ErrReplaced when a newer replaceable event
// already holds the slot.
func (e *Engine) SaveEvent(ev *nostr.Event) (bool, error) {
	e.mu.Lock()
	defer e.mu.Unlock()
	if e.closed {
		return false, ErrClosed
	}
	if e.bulk {
		if _, ok := e.bulkIDs[ev.ID]; ok {
			return false, ErrDuplicate
		}
	} else {
		dup, err := e.hasEvent(ev.ID[:])
		if err != nil {
			return false, err
		}
		if dup {
			return false, ErrDuplicate
		}
	}
	if e.isTombstoned(ev.ID[:]) {
		return false, ErrDeleted
	}
	if ev.IsReplaceable() || ev.IsAddressable() {
		if e.bulk {
			if oldSer, ok := e.bulkRep[bulkSlotKey(ev)]; ok {
				if old, gerr := e.getBySerial(oldSer); gerr == nil {
					if !eventWins(ev, old) {
						return false, ErrReplaced
					}
					e.removeEvent(old, oldSer)
				}
			}
		} else {
			oldSer, ok, err := e.currentReplacement(ev)
			if err != nil {
				return false, err
			}
			if ok {
				old, gerr := e.getBySerial(oldSer)
				if gerr == nil {
					if !eventWins(ev, old) {
						return false, ErrReplaced
					}
					e.removeEvent(old, oldSer)
				}
			}
		}
	}

	ser, err := e.w.Append(encodeEvent(ev))
	if err != nil {
		return false, err
	}
	if err := e.indexEvent(ev, ser); err != nil {
		return false, err
	}
	e.lastSer = ser
	if e.bulk {
		e.bulkIDs[ev.ID] = ser
		if ev.IsReplaceable() || ev.IsAddressable() {
			e.bulkRep[bulkSlotKey(ev)] = ser
		}
	}
	if ev.IsDeletion() {
		if err := e.applyDeletion(ev); err != nil {
			return true, err
		}
	}
	if e.bulk {
		return true, nil
	}
	e.saveN++
	if e.saveN >= 256 {
		e.saveN = 0
		if err := e.checkpointNow(); err != nil {
			return true, err
		}
	}
	return true, nil
}

// indexEvent writes every index record for one event.
func (e *Engine) indexEvent(ev *nostr.Event, ser uint64) error {
	idHash := Hash8(ev.ID[:])
	pubHash := Hash8(ev.PubKey[:])

	put := func(f *sortedFile, rec []byte) error { return f.Put(rec) }
	steps := []struct {
		f   *sortedFile
		rec []byte
	}{
		{e.eid, makeEid(idHash, ser)},
		{e.sei, makeSeiRec(ser, ev.ID[:])},
		{e.fpc, makeFpc(ser, ev.ID[:], pubHash[:], ev.CreatedAt)},
		{e.ca, makeCA(pCA, ev.CreatedAt, ser, caRecLen)},
		{e.kc, makeKC(ev.Kind, ev.CreatedAt, ser)},
		{e.pc, makePC(pubHash, ev.CreatedAt, ser)},
		{e.kpc, makeKPC(ev.Kind, pubHash, ev.CreatedAt, ser)},
	}
	for _, s := range steps {
		if err := put(s.f, s.rec); err != nil {
			return err
		}
	}

	var exp int64
	for _, tag := range ev.Tags {
		if len(tag) < 2 || len(tag[0]) != 1 {
			continue
		}
		tagKey := tag[0][0]
		vh := Hash8([]byte(tag[1]))
		if err := put(e.tc, makeTC(tagKey, vh, ev.CreatedAt, ser)); err != nil {
			return err
		}
		if err := put(e.tkc, makeTKC(ev.Kind, tagKey, vh, ev.CreatedAt, ser)); err != nil {
			return err
		}
		if err := put(e.tpc, makeTPC(pubHash, tagKey, vh, ev.CreatedAt, ser)); err != nil {
			return err
		}
		if err := put(e.tkp, makeTKP(ev.Kind, pubHash, tagKey, vh, ev.CreatedAt, ser)); err != nil {
			return err
		}
	}
	if tag, ok := ev.Tags.First("expiration"); ok {
		if n, err := strconv.ParseInt(tag, 10, 64); err == nil && n > 0 {
			exp = n
			if err := put(e.exp, makeCA(pExp, exp, ser, expRecLen)); err != nil {
				return err
			}
		}
	}

	words := splitWords(ev.Content)
	if len(words) > 512 {
		words = words[:512]
	}
	for _, word := range words {
		h := Hash8([]byte(word))
		if err := put(e.wrd, makeWrd(h, ser)); err != nil {
			return err
		}
	}

	if ev.IsReplaceable() || ev.IsAddressable() {
		dHash := [HashLen]byte{}
		if ev.IsAddressable() {
			dHash = Hash8([]byte(ev.DTag()))
		}
		if err := put(e.rep, makeRep(ev.Kind, pubHash, dHash, ev.CreatedAt, ser)); err != nil {
			return err
		}
	}
	if e.bulk {
		return nil
	}
	if err := e.indexGraph(ev, ser); err != nil {
		return err
	}
	return e.backfillInboundETags(ev, ser)
}

// repKey returns the replacement slot key for an event.
func bulkSlotKey(ev *nostr.Event) string {
	kind, pubHash, dHash := repKey(ev)
	return bulkSlotKeyParts(kind, pubHash, dHash)
}

func bulkSlotKeyParts(kind int, pubHash, dHash [HashLen]byte) string {
	var key [2 + HashLen + HashLen]byte
	PutKind(key[:2], kind)
	copy(key[2:], pubHash[:])
	copy(key[2+HashLen:], dHash[:])
	return string(key[:])
}

// loadBulkRep seeds the in-memory replacement slots from the persisted rep
// index. Without it, an import into an existing store would not know which
// replaceable event it must displace.
// loadBulkIDs seeds the in-memory id map from live events, so a bulk import
// into an existing store does not need to sort the growing eid buffer on every
// duplicate check.
func (e *Engine) loadBulkIDs() error {
	e.bulkIDs = make(map[[32]byte]uint64)
	return e.sei.Scan(nil, nil, func(rec []byte) bool {
		var id [32]byte
		copy(id[:], rec[seiRecLen:])
		e.bulkIDs[id] = GetSerial(rec[3:])
		return true
	})
}

func (e *Engine) loadBulkRep() error {
	e.bulkRep = make(map[string]uint64)
	latestTS := make(map[string]int64)
	return e.rep.Scan(nil, nil, func(rec []byte) bool {
		kind := GetKind(rec[3:5])
		var pubHash, dHash [HashLen]byte
		copy(pubHash[:], rec[5:5+HashLen])
		copy(dHash[:], rec[5+HashLen:5+2*HashLen])
		ts := GetTimestamp(rec[5+2*HashLen:])
		ser := GetSerial(rec[len(rec)-SerialLen:])
		key := bulkSlotKeyParts(kind, pubHash, dHash)
		if oldTS, ok := latestTS[key]; !ok || ts > oldTS {
			latestTS[key] = ts
			e.bulkRep[key] = ser
		}
		return true
	})
}

func repKey(ev *nostr.Event) (kind int, pubHash, dHash [HashLen]byte) {
	pubHash = Hash8(ev.PubKey[:])
	if ev.IsAddressable() {
		dHash = Hash8([]byte(ev.DTag()))
	}
	return ev.Kind, pubHash, dHash
}

// currentReplacement returns the serial of the newest event in the event's
// replacement slot.
func (e *Engine) currentReplacement(ev *nostr.Event) (uint64, bool, error) {
	kind, pubHash, dHash := repKey(ev)
	prefix := makeRepPrefix(kind, pubHash, dHash)
	end := prefixEnd(prefix)
	var last []byte
	err := e.rep.Scan(prefix, end, func(rec []byte) bool {
		last = rec
		return true
	})
	if err != nil {
		return 0, false, err
	}
	if last == nil {
		return 0, false, nil
	}
	return GetSerial(last[len(last)-SerialLen:]), true, nil
}

// applyReplacement is the replay-time equivalent of the replacement check in
// SaveEvent: it drops an older event when a newer one arrives.
func (e *Engine) applyReplacement(ev *nostr.Event) (bool, error) {
	if !ev.IsReplaceable() && !ev.IsAddressable() {
		return true, nil
	}
	var oldSer uint64
	var ok bool
	if e.bulk {
		oldSer, ok = e.bulkRep[bulkSlotKey(ev)]
	} else {
		var err error
		oldSer, ok, err = e.currentReplacement(ev)
		if err != nil {
			return false, err
		}
	}
	if !ok {
		return true, nil
	}
	old, err := e.getBySerial(oldSer)
	if err != nil {
		return true, nil
	}
	if !eventWins(ev, old) {
		return false, nil
	}
	e.removeEvent(old, oldSer)
	return true, nil
}

// eventWins reports whether a should replace b: newer created_at wins, and on
// a tie the lexicographically lowest id wins (NIP-01).
func eventWins(a, b *nostr.Event) bool {
	if a.CreatedAt != b.CreatedAt {
		return a.CreatedAt > b.CreatedAt
	}
	return bytes.Compare(a.ID[:], b.ID[:]) < 0
}

// removeEvent drops an event from every index. The WAL bytes stay; the event
// becomes unreachable and a rebuild replays replacement/deletion semantics.
func (e *Engine) removeEvent(ev *nostr.Event, ser uint64) {
	if e.bulk {
		e.removedSer[ser] = struct{}{}
		delete(e.bulkIDs, ev.ID)
		if ev.IsReplaceable() || ev.IsAddressable() {
			delete(e.bulkRep, bulkSlotKey(ev))
		}
		return
	}
	idHash := Hash8(ev.ID[:])
	pubHash := Hash8(ev.PubKey[:])
	del := func(f *sortedFile, rec []byte) { f.Delete(rec) }

	del(e.eid, makeEid(idHash, ser))
	del(e.sei, makeSei(ser))
	del(e.fpc, makeFpc(ser, ev.ID[:], pubHash[:], ev.CreatedAt))
	del(e.ca, makeCA(pCA, ev.CreatedAt, ser, caRecLen))
	del(e.kc, makeKC(ev.Kind, ev.CreatedAt, ser))
	del(e.pc, makePC(pubHash, ev.CreatedAt, ser))
	del(e.kpc, makeKPC(ev.Kind, pubHash, ev.CreatedAt, ser))

	for _, tag := range ev.Tags {
		if len(tag) < 2 || len(tag[0]) != 1 {
			continue
		}
		tagKey := tag[0][0]
		vh := Hash8([]byte(tag[1]))
		del(e.tc, makeTC(tagKey, vh, ev.CreatedAt, ser))
		del(e.tkc, makeTKC(ev.Kind, tagKey, vh, ev.CreatedAt, ser))
		del(e.tpc, makeTPC(pubHash, tagKey, vh, ev.CreatedAt, ser))
		del(e.tkp, makeTKP(ev.Kind, pubHash, tagKey, vh, ev.CreatedAt, ser))
	}
	if exp := ev.Expiration(); exp > 0 {
		del(e.exp, makeCA(pExp, exp, ser, expRecLen))
	}
	for _, word := range splitWords(ev.Content) {
		h := Hash8([]byte(word))
		del(e.wrd, makeWrd(h, ser))
	}
	if ev.IsReplaceable() || ev.IsAddressable() {
		kind, ph, dh := repKey(ev)
		del(e.rep, makeRep(kind, ph, dh, ev.CreatedAt, ser))
		if e.bulk {
			delete(e.bulkRep, bulkSlotKey(ev))
		}
	}
	if e.bulk {
		delete(e.bulkIDs, ev.ID)
	}
	if !e.bulk {
		e.removeGraph(ev, ser)
	}
}

// tombstone records a permanent deletion so that a WAL replay cannot bring
// the event back.
func (e *Engine) serialRemoved(ser uint64) bool {
	_, ok := e.removedSer[ser]
	return ok
}

func (e *Engine) tombstone(id []byte) error {
	return e.del.Put(makeDel(id))
}

func (e *Engine) isTombstoned(id []byte) bool {
	_, ok, _ := e.del.Get(makeDel(id))
	return ok
}

// applyDeletion implements NIP-09: the author of the deletion event may remove
// their own events, by id (e tag) or by address (a tag).
func (e *Engine) applyDeletion(ev *nostr.Event) error {
	for _, hexID := range ev.Tags.Values("e") {
		var id [32]byte
		if nostr.DecodeRaw(hexID, id[:]) != nil {
			continue
		}
		target, ser, err := e.lookup(id[:])
		if err != nil || target == nil {
			continue
		}
		if !bytes.Equal(target.PubKey[:], ev.PubKey[:]) {
			continue
		}
		if err := e.tombstone(id[:]); err != nil {
			return err
		}
		e.removeEvent(target, ser)
	}
	for _, addr := range ev.Tags.Values("a") {
		kind, pub, d, ok := parseAddress(addr)
		if !ok || !bytes.Equal(pub[:], ev.PubKey[:]) {
			continue
		}
		probe := &nostr.Event{Kind: kind}
		copy(probe.PubKey[:], pub[:])
		probe.Tags = nostr.Tags{{"d", d}}
		ser, found, err := e.currentReplacement(probe)
		if err != nil || !found {
			continue
		}
		target, err := e.getBySerial(ser)
		if err != nil || target == nil {
			continue
		}
		if err := e.tombstone(target.ID[:]); err != nil {
			return err
		}
		e.removeEvent(target, ser)
	}
	return nil
}

func parseAddress(s string) (kind int, pub [32]byte, d string, ok bool) {
	parts := strings.SplitN(s, ":", 3)
	if len(parts) != 3 {
		return 0, pub, "", false
	}
	k, err := strconv.Atoi(parts[0])
	if err != nil || k < 0 || k > nostr.MaxKind {
		return 0, pub, "", false
	}
	if nostr.DecodeRaw(parts[1], pub[:]) != nil {
		return 0, pub, "", false
	}
	return k, pub, parts[2], true
}

// hasEvent reports whether id is stored, filtering index hash collisions by
// comparing the full id held in the sei record.
func (e *Engine) hasEvent(id []byte) (bool, error) {
	rec, ok, err := e.eid.Get(makeEidPrefix(Hash8(id)))
	if err != nil || !ok {
		return false, err
	}
	ser := GetSerial(rec[3+HashLen:])
	seiRec, ok, err := e.sei.Get(makeSei(ser))
	if err != nil || !ok {
		return false, err
	}
	return bytes.Equal(seiRec[seiRecLen:seiRecLen+32], id), nil
}

// serialByID returns the serial for a live event id without decoding the
// event. Graph construction uses this for e-tag targets; decoding every target
// from the WAL turned a graph rebuild into a random-read workload.
func (e *Engine) serialByID(id []byte) (uint64, bool, error) {
	if e.bulk {
		var key [32]byte
		copy(key[:], id)
		ser, ok := e.bulkIDs[key]
		return ser, ok, nil
	}
	rec, ok, err := e.eid.Get(makeEidPrefix(Hash8(id)))
	if err != nil || !ok {
		return 0, false, err
	}
	ser := GetSerial(rec[3+HashLen:])
	seiRec, ok, err := e.sei.Get(makeSei(ser))
	if err != nil || !ok || len(seiRec) < seiRecLen+seiValLen {
		return 0, false, err
	}
	if !bytes.Equal(seiRec[seiRecLen:seiRecLen+seiValLen], id) {
		return 0, false, nil
	}
	return ser, true, nil
}

func (e *Engine) lookup(id []byte) (*nostr.Event, uint64, error) {
	if e.bulk {
		var key [32]byte
		copy(key[:], id)
		ser, ok := e.bulkIDs[key]
		if !ok {
			return nil, 0, ErrNotFound
		}
		ev, err := e.getBySerial(ser)
		if err != nil {
			return nil, 0, err
		}
		if !bytes.Equal(ev.ID[:], id) {
			return nil, 0, ErrNotFound
		}
		return ev, ser, nil
	}
	rec, ok, err := e.eid.Get(makeEidPrefix(Hash8(id)))
	if err != nil || !ok {
		return nil, 0, ErrNotFound
	}
	ser := GetSerial(rec[3+HashLen:])
	ev, err := e.getBySerial(ser)
	if err != nil {
		return nil, 0, err
	}
	if !bytes.Equal(ev.ID[:], id) {
		return nil, 0, ErrNotFound
	}
	return ev, ser, nil
}

func (e *Engine) getBySerial(ser uint64) (*nostr.Event, error) {
	data, err := e.w.Read(ser)
	if err != nil {
		return nil, err
	}
	return decodeEvent(data)
}

// GetByID returns a stored event by id.
func (e *Engine) GetByID(id []byte) (*nostr.Event, error) {
	e.mu.Lock()
	defer e.mu.Unlock()
	if e.isTombstoned(id) {
		return nil, ErrNotFound
	}
	ev, _, err := e.lookup(id)
	return ev, err
}

// GetBySerial returns a stored event by WAL serial.
func (e *Engine) GetBySerial(ser uint64) (*nostr.Event, error) {
	e.mu.Lock()
	defer e.mu.Unlock()
	ev, err := e.getBySerial(ser)
	if err != nil {
		return nil, err
	}
	if e.isTombstoned(ev.ID[:]) {
		return nil, ErrNotFound
	}
	return ev, nil
}

// Delete removes an event if byPub is its author. It returns false when the
// event does not exist or the author does not match.
func (e *Engine) Delete(id, byPub []byte) (bool, error) {
	e.mu.Lock()
	defer e.mu.Unlock()
	ev, ser, err := e.lookup(id)
	if err != nil {
		return false, nil
	}
	if !bytes.Equal(ev.PubKey[:], byPub) {
		return false, nil
	}
	if err := e.tombstone(id); err != nil {
		return false, err
	}
	e.removeEvent(ev, ser)
	return true, nil
}

// PurgeExpired removes NIP-40 events whose expiration is at or before now.
func (e *Engine) PurgeExpired(now int64) (int, error) {
	e.mu.Lock()
	defer e.mu.Unlock()
	if err := e.flushPending(); err != nil {
		return 0, err
	}
	start := makeCA(pExp, 0, 0, expRecLen)
	var end []byte
	if now >= math.MaxInt64-1 {
		end = prefixEnd([]byte{'e', 'x', 'p'})
	} else {
		end = makeCA(pExp, now+1, 0, expRecLen)
	}
	var serials []uint64
	if err := e.exp.Scan(start, end, func(rec []byte) bool {
		serials = append(serials, GetSerial(rec[len(rec)-SerialLen:]))
		return true
	}); err != nil {
		return 0, err
	}
	n := 0
	for _, ser := range serials {
		ev, err := e.getBySerial(ser)
		if err != nil {
			continue
		}
		exp := ev.Expiration()
		if exp == 0 || exp > now {
			continue
		}
		if err := e.tombstone(ev.ID[:]); err != nil {
			return n, err
		}
		e.removeEvent(ev, ser)
		n++
	}
	return n, nil
}

// Query returns matching events, newest first, capped by the filter limit and
// the engine's MaxResults.
func (e *Engine) Query(f *nostr.Filter) ([]*nostr.Event, error) {
	if f.Graph != nil {
		return nil, ErrGraph
	}
	e.mu.Lock()
	defer e.mu.Unlock()
	if e.closed {
		return nil, ErrClosed
	}
	if err := e.flushPending(); err != nil {
		return nil, fmt.Errorf("store: flush: %w", err)
	}
	cands, err := e.candidates(f)
	if err != nil {
		return nil, fmt.Errorf("store: candidates: %w", err)
	}
	limit := f.Limit
	if limit <= 0 || limit > e.maxResults {
		limit = e.maxResults
	}
	now := time.Now().Unix()

	// Index scans already carry created_at. The indexes are ordered by
	// (prefix, created_at), so sorting the candidate list by timestamp and
	// loading only until the limit is reached avoids reading every matching
	// event from the WAL. A broad filter such as kinds:[1] used to read and
	// decode the entire kind before returning one result.
	timed := true
	for i := range cands {
		if !cands[i].hasTS {
			timed = false
			break
		}
	}
	if timed {
		sort.Slice(cands, func(i, j int) bool {
			if cands[i].ts != cands[j].ts {
				return cands[i].ts > cands[j].ts
			}
			return cands[i].ser > cands[j].ser
		})
		out := make([]*nostr.Event, 0, limit)
		for _, c := range cands {
			ev, err := e.getBySerial(c.ser)
			if err != nil {
				continue
			}
			if e.isTombstoned(ev.ID[:]) {
				continue
			}
			if exp := ev.Expiration(); exp != 0 && exp <= now {
				continue
			}
			if !f.Matches(ev) {
				continue
			}
			out = append(out, ev)
			if len(out) >= limit {
				break
			}
		}
		return out, nil
	}

	// ID and full-text searches do not return a timestamp with their
	// candidates. There are usually few of those, so fall back to loading and
	// sorting the whole candidate set.
	out := make([]*nostr.Event, 0, len(cands))
	for _, c := range cands {
		ev, err := e.getBySerial(c.ser)
		if err != nil {
			continue
		}
		if e.isTombstoned(ev.ID[:]) {
			continue
		}
		if exp := ev.Expiration(); exp != 0 && exp <= now {
			continue
		}
		if !f.Matches(ev) {
			continue
		}
		out = append(out, ev)
	}
	sortEvents(out)
	if len(out) > limit {
		out = out[:limit]
	}
	return out, nil
}

// Count returns the number of matching events and whether the value is an
// approximation (the engine stopped at its result cap).
func (e *Engine) Count(f *nostr.Filter) (int, bool, error) {
	if f.Graph != nil {
		return 0, false, ErrGraph
	}
	e.mu.Lock()
	defer e.mu.Unlock()
	if e.closed {
		return 0, false, ErrClosed
	}
	if err := e.flushPending(); err != nil {
		return 0, false, fmt.Errorf("store: flush: %w", err)
	}
	cands, err := e.candidates(f)
	if err != nil {
		return 0, false, fmt.Errorf("store: candidates: %w", err)
	}
	now := time.Now().Unix()
	n := 0
	for _, c := range cands {
		ev, err := e.getBySerial(c.ser)
		if err != nil || e.isTombstoned(ev.ID[:]) {
			continue
		}
		if exp := ev.Expiration(); exp != 0 && exp <= now {
			continue
		}
		if !f.Matches(ev) {
			continue
		}
		n++
		if n >= e.maxResults {
			return n, true, nil
		}
	}
	return n, false, nil
}

func sortEvents(evs []*nostr.Event) {
	sort.SliceStable(evs, func(i, j int) bool {
		if evs[i].CreatedAt != evs[j].CreatedAt {
			return evs[i].CreatedAt > evs[j].CreatedAt
		}
		return bytes.Compare(evs[i].ID[:], evs[j].ID[:]) < 0
	})
}

// flushPending makes buffered index records visible to searches.
func (e *Engine) flushPending() error {
	for _, f := range e.indexes() {
		if err := f.QuickFlush(false); err != nil {
			return err
		}
	}
	return nil
}

// checkpointNow fsyncs the WAL and every index, then advances the checkpoint.
// Only after this returns may recovery start after the checkpoint position:
// before it, the sidecar writes are not guaranteed to have reached disk.
func (e *Engine) checkpointNow() error {
	if err := e.w.Sync(); err != nil {
		return err
	}
	for _, f := range e.indexes() {
		if err := f.QuickFlush(true); err != nil {
			return err
		}
	}
	if err := e.ckpt.Set(e.w.Pos(), e.nextPubSer); err != nil {
		return err
	}
	return e.ckpt.Sync()
}

// Flush writes everything durably and merges each index run into its main
// file. It is safe to call at any time.
func (e *Engine) Flush() error {
	e.mu.Lock()
	defer e.mu.Unlock()
	if e.closed {
		return ErrClosed
	}
	return e.flushLocked()
}

func (e *Engine) flushLocked() error {
	if err := e.checkpointNow(); err != nil {
		return err
	}
	for _, f := range e.indexes() {
		if err := f.Flush(); err != nil {
			return err
		}
	}
	e.removedSer = nil
	return e.ckpt.Sync()
}

// Compact merges every index without forcing an additional checkpoint.
func (e *Engine) Compact() error {
	e.mu.Lock()
	defer e.mu.Unlock()
	if e.closed {
		return ErrClosed
	}
	for _, f := range e.indexes() {
		if err := f.Flush(); err != nil {
			return err
		}
	}
	return nil
}

// Close flushes and closes the engine.
func (e *Engine) Close() error {
	e.mu.Lock()
	defer e.mu.Unlock()
	if e.closed {
		return nil
	}
	e.closed = true
	var first error
	if err := e.w.Sync(); err != nil && first == nil {
		first = err
	}
	for _, f := range e.indexes() {
		if err := f.Flush(); err != nil && first == nil {
			first = err
		}
	}
	if err := e.ckpt.Set(e.w.Pos(), e.nextPubSer); err != nil && first == nil {
		first = err
	}
	if err := e.ckpt.Sync(); err != nil && first == nil {
		first = err
	}
	for _, f := range e.indexes() {
		if err := f.Close(); err != nil && first == nil {
			first = err
		}
	}
	if err := e.w.Close(); err != nil && first == nil {
		first = err
	}
	if err := e.ckpt.Close(); err != nil && first == nil {
		first = err
	}
	return first
}

// --- query planning ---

// candidate is a serial plus the created_at carried by the index that
// produced it. Records from eid/sei/wrd have no timestamp; callers fall back
// to loading those events.
type candidate struct {
	ser   uint64
	ts    int64
	hasTS bool
}

func (e *Engine) candidates(f *nostr.Filter) ([]candidate, error) {
	if len(f.IDs) > 0 {
		return e.candidatesByID(f.IDs)
	}
	if f.Search != "" {
		return e.candidatesBySearch(f.Search)
	}

	since := f.Since
	until := f.Until
	if until == 0 {
		until = math.MaxInt64
	}
	hasKinds := len(f.Kinds) > 0
	hasAuthors := len(f.Authors) > 0
	hasTags := len(f.Tags) > 0

	var serials []candidate
	seen := make(map[uint64]struct{})
	add := func(ser uint64, ts int64, hasTS bool) {
		if _, ok := seen[ser]; ok {
			return
		}
		seen[ser] = struct{}{}
		serials = append(serials, candidate{ser: ser, ts: ts, hasTS: hasTS})
	}
	collect := func(idx *sortedFile, start, end []byte, tsOff int) error {
		return idx.Scan(start, end, func(rec []byte) bool {
			ser := GetSerial(rec[len(rec)-SerialLen:])
			if tsOff >= 0 && tsOff+8 <= len(rec) {
				add(ser, GetTimestamp(rec[tsOff:]), true)
			} else {
				add(ser, 0, false)
			}
			return true
		})
	}

	// Range end: exclusive at the first timestamp after until. When until is
	// the maximum, the end is the successor of the fixed key prefix.
	rngEnd := func(start []byte, prefixLen int, build func(int64) []byte) []byte {
		if until >= math.MaxInt64-1 {
			return prefixEnd(start[:prefixLen])
		}
		return build(until + 1)
	}

	switch {
	case hasTags && hasKinds && hasAuthors:
		if err := e.forEachTag(f, func(tagKey byte, vh [HashLen]byte) error {
			for _, kind := range f.Kinds {
				for _, author := range f.Authors {
					ph := authorHash(author)
					start := makeTKP(kind, ph, tagKey, vh, since, 0)
					end := rngEnd(start, 22, func(ts int64) []byte {
						return makeTKP(kind, ph, tagKey, vh, ts, 0)
					})
					if err := collect(e.tkp, start, end, 22); err != nil {
						return err
					}
				}
			}
			return nil
		}); err != nil {
			return nil, err
		}
	case hasTags && hasKinds:
		if err := e.forEachTag(f, func(tagKey byte, vh [HashLen]byte) error {
			for _, kind := range f.Kinds {
				start := makeTKC(kind, tagKey, vh, since, 0)
				end := rngEnd(start, 14, func(ts int64) []byte {
					return makeTKC(kind, tagKey, vh, ts, 0)
				})
				if err := collect(e.tkc, start, end, 14); err != nil {
					return err
				}
			}
			return nil
		}); err != nil {
			return nil, err
		}
	case hasTags && hasAuthors:
		if err := e.forEachTag(f, func(tagKey byte, vh [HashLen]byte) error {
			for _, author := range f.Authors {
				ph := authorHash(author)
				start := makeTPC(ph, tagKey, vh, since, 0)
				end := rngEnd(start, 20, func(ts int64) []byte {
					return makeTPC(ph, tagKey, vh, ts, 0)
				})
				if err := collect(e.tpc, start, end, 20); err != nil {
					return err
				}
			}
			return nil
		}); err != nil {
			return nil, err
		}
	case hasTags:
		if err := e.forEachTag(f, func(tagKey byte, vh [HashLen]byte) error {
			start := makeTC(tagKey, vh, since, 0)
			end := rngEnd(start, 12, func(ts int64) []byte {
				return makeTC(tagKey, vh, ts, 0)
			})
			return collect(e.tc, start, end, 12)
		}); err != nil {
			return nil, err
		}
	case hasKinds && hasAuthors:
		for _, kind := range f.Kinds {
			for _, author := range f.Authors {
				ph := authorHash(author)
				start := makeKPC(kind, ph, since, 0)
				end := rngEnd(start, 13, func(ts int64) []byte {
					return makeKPC(kind, ph, ts, 0)
				})
				if err := collect(e.kpc, start, end, 13); err != nil {
					return nil, err
				}
			}
		}
	case hasKinds:
		for _, kind := range f.Kinds {
			start := makeKC(kind, since, 0)
			end := rngEnd(start, 5, func(ts int64) []byte {
				return makeKC(kind, ts, 0)
			})
			if err := collect(e.kc, start, end, 5); err != nil {
				return nil, err
			}
		}
	case hasAuthors:
		for _, author := range f.Authors {
			ph := authorHash(author)
			start := makePC(ph, since, 0)
			end := rngEnd(start, 11, func(ts int64) []byte {
				return makePC(ph, ts, 0)
			})
			if err := collect(e.pc, start, end, 11); err != nil {
				return nil, err
			}
		}
	default:
		start := makeCA(pCA, since, 0, caRecLen)
		end := rngEnd(start, 3, func(ts int64) []byte {
			return makeCA(pCA, ts, 0, caRecLen)
		})
		if err := collect(e.ca, start, end, 3); err != nil {
			return nil, err
		}
	}
	return serials, nil
}

func (e *Engine) candidatesByID(ids []string) ([]candidate, error) {
	var out []candidate
	seen := make(map[uint64]struct{})
	for _, h := range ids {
		var id [32]byte
		if nostr.DecodeRaw(h, id[:]) != nil {
			continue
		}
		ev, ser, err := e.lookup(id[:])
		if err != nil || ev == nil {
			continue
		}
		if _, ok := seen[ser]; ok {
			continue
		}
		seen[ser] = struct{}{}
		out = append(out, candidate{ser: ser})
	}
	return out, nil
}

func (e *Engine) candidatesBySearch(query string) ([]candidate, error) {
	words := splitWords(query)
	if len(words) == 0 {
		return nil, nil
	}
	// Deduplicate query words.
	uniq := words[:0]
	seenWord := make(map[string]struct{}, len(words))
	for _, w := range words {
		if _, ok := seenWord[w]; ok {
			continue
		}
		seenWord[w] = struct{}{}
		uniq = append(uniq, w)
	}

	var sets [][]uint64
	for _, word := range uniq {
		h := Hash8([]byte(word))
		start := makeWrd(h, 0)
		end := prefixEnd(start[:3+HashLen])
		var serials []uint64
		if err := e.wrd.Scan(start, end, func(rec []byte) bool {
			serials = append(serials, GetSerial(rec[len(rec)-SerialLen:]))
			return true
		}); err != nil {
			return nil, err
		}
		if len(serials) == 0 {
			return nil, nil
		}
		sets = append(sets, serials)
	}

	result := sets[0]
	for _, next := range sets[1:] {
		set := make(map[uint64]struct{}, len(next))
		for _, s := range next {
			set[s] = struct{}{}
		}
		var keep []uint64
		for _, s := range result {
			if _, ok := set[s]; ok {
				keep = append(keep, s)
			}
		}
		result = keep
		if len(result) == 0 {
			return nil, nil
		}
	}
	out := make([]candidate, len(result))
	for i, ser := range result {
		out[i] = candidate{ser: ser}
	}
	return out, nil
}

// authorHash hashes the raw 32 pubkey bytes of a hex author, matching the way
// the pc/kpc/tpc/tkp indexes hash an event's PubKey field.
func authorHash(hexAuthor string) (h [HashLen]byte) {
	var raw [32]byte
	if nostr.DecodeRaw(hexAuthor, raw[:]) != nil {
		return Hash8([]byte(hexAuthor))
	}
	return Hash8(raw[:])
}

func (e *Engine) forEachTag(f *nostr.Filter, fn func(tagKey byte, vh [HashLen]byte) error) error {
	for name, values := range f.Tags {
		if len(name) != 1 {
			continue
		}
		for _, value := range values {
			if err := fn(name[0], Hash8([]byte(value))); err != nil {
				return err
			}
		}
	}
	return nil
}

// splitWords lowercases ASCII runs of letters and digits and keeps runs of at
// least three characters, matching the NIP-50 tokenizer used by morly.
func splitWords(s string) []string {
	var out []string
	start := -1
	buf := make([]byte, 0, 32)
	flush := func(end int) {
		if start < 0 {
			return
		}
		if len(buf) >= 3 {
			out = append(out, string(buf))
		}
		buf = buf[:0]
		start = -1
	}
	for i := 0; i < len(s); i++ {
		c := s[i]
		switch {
		case c >= 'a' && c <= 'z', c >= '0' && c <= '9':
			if start < 0 {
				start = i
			}
			buf = append(buf, c)
		case c >= 'A' && c <= 'Z':
			if start < 0 {
				start = i
			}
			buf = append(buf, c+32)
		default:
			flush(i)
		}
	}
	flush(len(s))
	return out
}