func (m *EQMem) queueUnsafeInsertTask(ql *qLock, t *entroq.Task) func() {
	ql.heap.PushItem(newItem(ql.queue, t.ID, t.At))
	ql.tasks.Set(t.ID, t)
	return func() {
		m.qByID[t.ID] = t.Queue
	}
}
queueUnsafeInsertTask performs queue-level operations on a task, then
returns a function to call under global lock to finish the job.
package eqmem
import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"math/rand"
	"sort"
	"strings"
	"sync"
	"time"
	"github.com/google/uuid"
	"github.com/shiblon/entroq"
	"github.com/shiblon/entroq/subq"
	"github.com/shiblon/stuffedio/wal"
)
type EQMem struct {
	sync.Mutex
	nw entroq.NotifyWaiter
	queues map[string]*taskQueue
	qByID map[uuid.UUID]string
	locks map[string]*qLock
	journal *wal.WAL
	journalDir      string
	maxJournalBytes int64
	maxJournalItems int
	outputSnapshot bool
}
type qLock struct {
	sync.Mutex
	queue string
	heap  *claimHeap
	tasks *taskQueue
}
func lock(l sync.Locker) func() {
	l.Lock()
	return l.Unlock
}
func un(f func()) {
	f()
}
func Opener(opts ...Option) entroq.BackendOpener {
	return func(ctx context.Context) (entroq.Backend, error) {
		back, err := New(ctx, opts...)
		return back, err
	}
}
type Option func(*EQMem)
func WithJournal(dir string) Option {
	return func(m *EQMem) {
		m.journalDir = dir
	}
}
func WithMaxJournalBytes(max int64) Option {
	return func(m *EQMem) {
		if max <= 0 {
			return
		}
		m.maxJournalBytes = max
	}
}
func WithMaxJournalItems(max int) Option {
	return func(m *EQMem) {
		if max <= 0 {
			return
		}
		m.maxJournalItems = max
	}
}
func withOutputSnapshot() Option {
	return func(m *EQMem) {
		m.outputSnapshot = true
	}
}
func New(ctx context.Context, opts ...Option) (*EQMem, error) {
	m := &EQMem{
		nw:     subq.New(),
		queues: make(map[string]*taskQueue),
		qByID:  make(map[uuid.UUID]string),
		locks:  make(map[string]*qLock),
	}
	for _, opt := range opts {
		opt(m)
	}
	if m.journalDir != "" {
		walOpts := []wal.Option{
			wal.WithMaxJournalBytes(m.maxJournalBytes),
			wal.WithMaxJournalIndices(m.maxJournalItems),
			wal.WithAllowWrite(!m.outputSnapshot),
			wal.WithExcludeLiveJournal(m.outputSnapshot),
			wal.WithSnapshotLoaderFunc(func(ctx context.Context, b []byte) error {
				task := new(entroq.Task)
				if err := json.Unmarshal(b, task); err != nil {
					return fmt.Errorf("eqmem load task: %w", err)
				}
				var ql *qLock
				func() {
					defer un(lock(m))
					m.unsafeEnsureQueue(task.Queue)
					ql = m.locks[task.Queue]
				}()
				defer un(lock(ql))
				finish := m.queueUnsafeInsertTask(ql, task)
				defer un(lock(m))
				finish()
				return nil
			}),
			wal.WithJournalPlayerFunc(func(ctx context.Context, b []byte) error {
				mod := new(entroq.Modification)
				if err := json.Unmarshal(b, mod); err != nil {
					return fmt.Errorf("eqmem play mod: %w", err)
				}
				for _, chg := range mod.Changes {
					chg.Version--
				}
				if _, _, err := m.modifyImpl(ctx, mod, true); err != nil {
					return fmt.Errorf("eqmem play mod: %w", err)
				}
				return nil
			}),
		}
		var err error
		if m.journal, err = wal.Open(ctx, m.journalDir, walOpts...); err != nil {
			return nil, fmt.Errorf("open WAL: %w", err)
		}
		if m.outputSnapshot {
			if !m.journal.SnapshotUseful() {
				log.Printf("Snapshot requested, but not useful: empty, or frozen journals already collapsed")
				return m, nil
			}
			if _, err := m.journal.CreateSnapshot(m.makeSnapshot); err != nil {
				return nil, fmt.Errorf("output snapshot: %w", err)
			}
		}
	}
	return m, nil
}
func TakeSnapshot(ctx context.Context, journalDir string, cleanup bool) error {
	m, err := New(ctx, WithJournal(journalDir), withOutputSnapshot())
	if err != nil {
		return fmt.Errorf("load for snapshot: %w", err)
	}
	defer m.Close()
	if cleanup {
		if err := wal.Cleanup(journalDir); err != nil {
			return fmt.Errorf("snapshot cleanup: %w", err)
		}
	}
	return nil
}
func (m *EQMem) makeSnapshot(a wal.ValueAdder) error {
	var err error
	for _, ts := range m.queues {
		ts.Range(func(_ uuid.UUID, t *entroq.Task) bool {
			var b []byte
			if b, err = json.Marshal(t); err != nil {
				err = fmt.Errorf("marshal for snapshot: %w", err)
				return false
			}
			if err = a.AddValue(b); err != nil {
				err = fmt.Errorf("add value: %w", err)
				return false
			}
			return true
		})
	}
	return err
}
func (m *EQMem) claimPrep(cq *entroq.ClaimQuery) []*qLock {
	defer un(lock(m))
	var qls []*qLock
	for _, q := range cq.Queues {
		if ql, ok := m.locks[q]; ok {
			qls = append(qls, ql)
		}
	}
	return qls
}
func (m *EQMem) mustTryClaimOne(ql *qLock, now time.Time, cq *entroq.ClaimQuery) *entroq.Task {
	defer un(lock(ql))
	item := ql.heap.RandomAvailable(now)
	if item == nil {
		return nil
	}
	qts, ok := m.queueTasks(ql.queue)
	if !ok {
		log.Fatalf("Inconsistent internal state: could not find queue %q after finding a claimable task in it", ql.queue)
	}
	newAt := now.Add(cq.Duration)
	ql.heap.UpdateItem(item, newAt)
	var found *entroq.Task
	if err := qts.Update(item.id, func(t *entroq.Task) *entroq.Task {
		t = t.Copy()  
		t.At = newAt
		t.Claimant = cq.Claimant
		t.Version++
		t.Claims++
		t.Modified = now
		found = t
		return t
	}); err != nil {
		log.Fatalf("Inconsistent internal state: could not update task in %q after claim started", ql.queue)
	}
	if m.journal != nil {
		mod := &entroq.Modification{
			Claimant: cq.Claimant,
			Changes:  []*entroq.Task{found},
		}
		b, err := json.Marshal(mod)
		if err != nil {
			log.Fatalf("Inconsistent internal state: updated task but couldn't marshal JSON: %v", err)
		}
		if err := m.journal.Append(b); err != nil {
			log.Fatalf("Inconsistent internal state: updated task but couldn't write to journal: %v", err)
		}
	}
	return found
}
func (m *EQMem) Claim(ctx context.Context, cq *entroq.ClaimQuery) (*entroq.Task, error) {
	return entroq.WaitTryClaim(ctx, cq, m.TryClaim, m.nw)
}
func (m *EQMem) TryClaim(ctx context.Context, cq *entroq.ClaimQuery) (*entroq.Task, error) {
	qls := m.claimPrep(cq)
	if len(qls) == 0 {
		return nil, nil
	}
	now, err := m.Time(ctx)
	if err != nil {
		return nil, fmt.Errorf("eqmem claim time: %w", err)
	}
	rand.Shuffle(len(qls), func(i, j int) {
		qls[i], qls[j] = qls[j], qls[i]
	})
	for _, ql := range qls {
		if task := m.mustTryClaimOne(ql, now, cq); task != nil {
			return task, nil
		}
	}
	return nil, nil
}
func (m *EQMem) unsafeEnsureQueue(q string) {
	ts, ok := m.queues[q]
	if ts == nil || !ok {
		ts = newTaskQueue(q)
		m.queues[q] = ts
	}
	if ql, ok := m.locks[q]; ql == nil || !ok {
		m.locks[q] = &qLock{
			queue: q,
			heap:  newClaimHeap(),
			tasks: ts,
		}
	}
}
func (m *EQMem) unsafeCleanQueue(q string) {
	ts, ok := m.queues[q]
	if ok && ts.Len() == 0 {
		delete(m.queues, q)
		delete(m.locks, q)
	}
}
func ensureModQueues(mod *entroq.Modification, qByID map[uuid.UUID]string) {
	for _, d := range mod.Deletes {
		if d.Queue == "" {
			d.Queue = qByID[d.ID]
		}
	}
	for _, d := range mod.Depends {
		if d.Queue == "" {
			d.Queue = qByID[d.ID]
		}
	}
	for _, c := range mod.Changes {
		c.FromQueue = qByID[c.ID]
	}
}
func (m *EQMem) modPrep(mod *entroq.Modification) (locks []*qLock, misplacedInsIDs map[uuid.UUID]string) {
	defer un(lock(m))
	misplacedInsIDs = make(map[uuid.UUID]string)
	ensureModQueues(mod, m.qByID)
	queues := make(map[string]bool)
	for _, ins := range mod.Inserts {
		if ins.ID != uuid.Nil {
			if foundQueue, ok := m.qByID[ins.ID]; ok && foundQueue != ins.Queue {
				misplacedInsIDs[ins.ID] = foundQueue
				queues[foundQueue] = true
			}
		}
		queues[ins.Queue] = true
	}
	for _, c := range mod.Changes {
		queues[c.FromQueue] = true
		queues[c.Queue] = true
	}
	for _, d := range mod.Deletes {
		queues[d.Queue] = true
	}
	for _, d := range mod.Depends {
		queues[d.Queue] = true
	}
	for q := range queues {
		if q == "" {
			continue
		}
		m.unsafeEnsureQueue(q)
		locks = append(locks, m.locks[q])
	}
	sort.Slice(locks, func(i, j int) bool {
		return locks[i].queue < locks[j].queue
	})
	return locks, misplacedInsIDs
}