package spindle

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"runtime/debug"
	"sync"
	"time"

	"tangled.org/core/api/org_tangled"
	"tangled.org/core/jetstream"
)

const (
	ingestShardCount = 8
	ingestShardQueue = 256
)

var errIngestStopped = errors.New("the ingest shards stopped")

type ingestTicket interface {
	Done()
	Drop()
}

type ingestItem struct {
	ctx    context.Context
	e      *jetstream.Event
	ticket ingestTicket
}

// same repo always lands on the same shard, that's where ordering comes from
type ingestShards struct {
	apply  func(context.Context, *jetstream.Event) error
	queues [ingestShardCount]chan ingestItem
	wg     sync.WaitGroup

	// stop takes the write lock, which keeps any send from landing after it
	mu      sync.RWMutex
	stopped bool
	quit    chan struct{}
	quitOne sync.Once
}

func newIngestShards(ctx context.Context, apply func(context.Context, *jetstream.Event) error) *ingestShards {
	sh := &ingestShards{apply: apply, quit: make(chan struct{})}
	for i := range sh.queues {
		sh.queues[i] = make(chan ingestItem, ingestShardQueue)
		sh.wg.Go(func() { sh.run(ctx, sh.queues[i]) })
	}
	return sh
}

// declarations shard by the repo they name, which puts them behind its other events
func ingestShardKey(e *jetstream.Event) string {
	if e.Commit != nil && e.Commit.Collection == org_tangled.RepoDeclarationNSID && e.Commit.Operation != jetstream.CommitOperationDelete {
		var declaration org_tangled.RepoDeclaration
		if json.Unmarshal(e.Commit.Record, &declaration) == nil && declaration.Repo != "" {
			return declaration.Repo
		}
	}
	return e.Did
}

func ingestShardOf(key string) int {
	return int(fnv32a(key) % ingestShardCount)
}

func (sh *ingestShards) dispatch(ctx context.Context, e *jetstream.Event, ticket ingestTicket) error {
	sh.mu.RLock()
	defer sh.mu.RUnlock()
	if sh.stopped {
		ticket.Drop()
		return errIngestStopped
	}
	select {
	case sh.queues[ingestShardOf(ingestShardKey(e))] <- ingestItem{ctx: ctx, e: e, ticket: ticket}:
		return nil
	case <-ctx.Done():
		ticket.Drop()
		return ctx.Err()
	case <-sh.quit:
		ticket.Drop()
		return errIngestStopped
	}
}

func (sh *ingestShards) run(ctx context.Context, queue chan ingestItem) {
	for {
		select {
		case it := <-queue:
			// even after cancel, select can pick a queued event, check here
			if ctx.Err() != nil {
				it.ticket.Drop()
				continue
			}
			sh.handle(it)
		case <-ctx.Done():
			sh.quitOne.Do(func() { close(sh.quit) })
			sh.mu.Lock()
			sh.stopped = true
			sh.mu.Unlock()
			for {
				select {
				case it := <-queue:
					it.ticket.Drop()
				default:
					return
				}
			}
		}
	}
}

// without this a panic kills the process before the cursor moves, and every start dies on it again
func recovered(what string, fn func() error) (err error) {
	defer func() {
		if p := recover(); p != nil {
			slog.Error("handler panicked, skipping it", "what", what, "panic", p, "stack", string(debug.Stack()))
			err = fmt.Errorf("panic in %s: %v: %w", what, p, errHandlerPanicked)
		}
	}()
	return fn()
}

var errHandlerPanicked = errors.New("the handler panicked")

const ingestAttempts = 5

// failures are mostly a busy db, which makes a few retries worth it
var ingestRetryDelay = 100 * time.Millisecond

func (sh *ingestShards) handle(it ingestItem) {
	var err error
	for attempt := range ingestAttempts {
		err = recovered("ingest shard", func() error { return sh.apply(it.ctx, it.e) })
		if err == nil || errors.Is(err, errHandlerPanicked) || it.ctx.Err() != nil {
			break
		}
		if attempt < ingestAttempts-1 {
			select {
			case <-time.After(ingestRetryDelay << attempt):
			case <-it.ctx.Done():
			}
		}
	}
	// the cursor stays behind a failed event, which means the next start reads it again
	if err != nil && !errors.Is(err, errHandlerPanicked) && it.ctx.Err() == nil {
		slog.Error("giving up on an event, the cursor stays behind it", "did", it.e.Did, "err", err)
		it.ticket.Drop()
		return
	}
	finishTicket(it.ctx, it.ticket, err)
}

// on cancel, drop rather than done, or the cursor would move past it
func finishTicket(ctx context.Context, t ingestTicket, err error) {
	if err != nil && ctx.Err() != nil {
		t.Drop()
		return
	}
	t.Done()
}