package views

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"time"

	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"
)

// Emitter is an owner's process saying each change of its own states on its views family (ADR 0615 §5): it follows
// each state it is given from the revision its mark says it said up to, publishes one Event per change, and writes the
// mark once it has said every change the bus had handed it. A part watches only its own state (ADR 0596 §4): an owner
// follows its own buckets, never another's.
type Emitter struct {
	// Owner is the owner module, said in every event.
	Owner string
	// Subject is where an event of the given local name lands: the owner's ordinary event (process.Account.EventSubject).
	Subject func(event string) string
	// Publish sends one message to the shared events stream, its id the bus's dedup id (JetStreamPublish).
	Publish func(ctx context.Context, msg *nats.Msg, id string) error
	// Marks is the owner's state MarksState; nil says nothing is marked, and every start says from the bucket's end.
	Marks jetstream.KeyValue
	// TooLarge is told of a value too large for an event (said as a gap): the owner's finding naming the key. Nil
	// only logs it.
	TooLarge func(state, key string, size int)
	// Findings writes and clears the owner's finding of a change not said past UnsaidAfter (UnsaidKind, its own state
	// `findings`); nil says it in the log only.
	Findings FindingWriter
	// UnsaidAfter is how long a change may stay not said before it is the owner's finding; zero is Bound.
	UnsaidAfter time.Duration
	// Wait pauses before a change not said is said again (its attempt so far); it answers false when ctx ended. Nil is
	// a backoff from 250 ms doubling to 30 s.
	Wait func(ctx context.Context, attempt int) bool
	Logf func(string, ...any)
	Now  func() time.Time
}

// JetStreamPublish is Emitter.Publish through js, with the bus's dedup id.
func JetStreamPublish(js jetstream.JetStream) func(context.Context, *nats.Msg, string) error {
	return func(ctx context.Context, msg *nats.Msg, id string) error {
		pub, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
		defer cancel()
		_, err := js.PublishMsg(pub, msg, jetstream.WithMsgID(id))
		return err
	}
}

func (e Emitter) now() time.Time {
	if e.Now != nil {
		return e.Now().UTC()
	}
	return time.Now().UTC()
}

func (e Emitter) logf(format string, args ...any) {
	if e.Logf != nil {
		e.Logf(format, args...)
	}
}

// created is when the bucket's stream was made, and its fill now.
func created(ctx context.Context, kv jetstream.KeyValue) (time.Time, uint64, Fill, error) {
	st, err := kv.Status(ctx)
	if err != nil {
		return time.Time{}, 0, Fill{}, err
	}
	b, ok := st.(*jetstream.KeyValueBucketStatus)
	if !ok || b.StreamInfo() == nil {
		return time.Time{}, 0, Fill{}, fmt.Errorf("the bus answered no stream behind %s", st.Bucket())
	}
	info := b.StreamInfo()
	return info.Created.UTC(), info.State.LastSeq, Fill{Keys: uint64(info.State.NumSubjects), Bytes: info.State.Bytes, MaxBytes: info.Config.MaxBytes}, nil
}

// Start follows state's bucket kv: from the revision after its mark when the mark is of the same bucket (the same
// Created), otherwise from the bucket's end, writing a mark there first, so a reader's first view taken now misses
// nothing. It returns once the watch is set; the changes are said on a goroutine until ctx ends, and a watch that ends
// is sent on errs.
func (e Emitter) Start(ctx context.Context, state string, kv jetstream.KeyValue, errs chan<- error) error {
	made, last, fill, err := created(ctx, kv)
	if err != nil {
		return fmt.Errorf("the state %s cannot be followed for its views: %w", state, err)
	}
	from := last + 1
	if m, ok := e.mark(ctx, state); ok && m.Created.Equal(made) && m.Revision <= last {
		from = m.Revision + 1
	} else {
		e.writeMark(ctx, state, Mark{State: state, Revision: last, Created: made, Fill: fill})
	}
	w, err := kv.WatchAll(ctx, jetstream.ResumeFromRevision(from))
	if err != nil {
		return fmt.Errorf("the state %s cannot be watched for its views: %w", state, err)
	}
	fillNow := func() (Fill, error) {
		_, _, fill, err := created(ctx, kv)
		return fill, err
	}
	go func() {
		defer w.Stop()
		e.follow(ctx, state, made, w.Updates(), fillNow, errs)
	}()
	return nil
}

// follow says each change handed on updates in order, and marks once every change handed so far is said. A change
// that fails to be said is said again, the same change on a backoff (Wait), until it is said or ctx ends: the mark
// never moves past a change not said, so no reader misses one (design 25 §12.2).
func (e Emitter) follow(ctx context.Context, state string, made time.Time, updates <-chan jetstream.KeyValueEntry, fillNow func() (Fill, error), errs chan<- error) {
	said := uint64(0)
	raised := false
	// A finding a run before this one raised is not known to this one: the first change said clears it once, so a
	// restart never leaves it standing (the finding is about the state, its key the same).
	leftover := e.Findings != nil
	for {
		select {
		case <-ctx.Done():
			return
		case u, ok := <-updates:
			if !ok {
				if errs != nil {
					errs <- fmt.Errorf("the watch of %s for its views ended", state)
				}
				return
			}
			if u != nil {
				var first time.Time
				for attempt := 1; ; attempt++ {
					err := e.say(ctx, state, made, u)
					if err == nil {
						break
					}
					e.logf("views: %s.%s at %d is not said (try %d): %v; said again before the next change", state, u.Key(), u.Revision(), attempt, err)
					if first.IsZero() {
						first = e.now()
					}
					if !raised && e.now().Sub(first) >= e.unsaidAfter() {
						raised = e.raiseUnsaid(state, u, first, err)
					}
					if !e.wait(ctx, attempt) {
						// Not marked: the next start says it again from the mark, and the bus keeps it once by its id.
						return
					}
				}
				if raised || leftover {
					raised = !e.clearUnsaid(state)
					leftover = false
				}
				said = u.Revision()
			}
			// Marked once every change handed so far is said, so a busy state writes one mark per burst.
			if said > 0 && len(updates) == 0 {
				fill, err := fillNow()
				if err != nil {
					e.logf("views: the fill of %s is not read: %v", state, err)
				}
				e.writeMark(ctx, state, Mark{State: state, Revision: said, Created: made, Fill: fill})
				said = 0
			}
		}
	}
}

func (e Emitter) unsaidAfter() time.Duration {
	if e.UnsaidAfter > 0 {
		return e.UnsaidAfter
	}
	return Bound
}

// raiseUnsaid writes the owner's finding of u not said since first; it answers whether the finding stands.
func (e Emitter) raiseUnsaid(state string, u jetstream.KeyValueEntry, first time.Time, cause error) bool {
	e.logf("views: %s.%s at %d is not said since %s: the owner's finding %s", state, u.Key(), u.Revision(), first.Format(time.RFC3339), UnsaidKind)
	if e.Findings == nil {
		return false
	}
	if err := e.Findings.Put(UnsaidFinding(e.Owner, state, u.Key(), u.Revision(), first, cause.Error(), e.now())); err != nil {
		e.logf("views: the finding of %s's change not said could not be written: %v", state, err)
		return false
	}
	return true
}

// clearUnsaid clears the finding once the change is said; it answers whether it is cleared.
func (e Emitter) clearUnsaid(state string) bool {
	if err := e.Findings.Delete(UnsaidFinding(e.Owner, state, "", 0, time.Time{}, "", e.now())); err != nil {
		e.logf("views: the finding of %s's change not said could not be cleared: %v", state, err)
		return false
	}
	return true
}

// wait is the pause before the attempt+1'th try of a change not said: Wait when given, else a backoff from 250 ms
// doubling to 30 s. It answers false once ctx ends.
func (e Emitter) wait(ctx context.Context, attempt int) bool {
	if e.Wait != nil {
		return e.Wait(ctx, attempt)
	}
	d := 250 * time.Millisecond
	for i := 1; i < attempt && d < 30*time.Second; i++ {
		d *= 2
	}
	d = min(d, 30*time.Second)
	t := time.NewTimer(d)
	defer t.Stop()
	select {
	case <-ctx.Done():
		return false
	case <-t.C:
		return true
	}
}

// Event is the event saying u, a change of state made at made.
func (e Emitter) Event(state string, made time.Time, u jetstream.KeyValueEntry) Event {
	ev := Event{V: Version, Owner: e.Owner, State: state, Key: u.Key(), Revision: u.Revision(), Created: made, At: u.Created().UTC()}
	if u.Operation() != jetstream.KeyValuePut {
		ev.Deleted = true
		return ev
	}
	if n := len(u.Value()); n > MaxValueBytes {
		ev.Gap = true
		e.logf("views: %s.%s at %d holds %d bytes, more than an event carries: said as a gap, so a reader asks its first view", state, u.Key(), u.Revision(), n)
		if e.TooLarge != nil {
			e.TooLarge(state, u.Key(), n)
		}
		return ev
	}
	ev.Value = json.RawMessage(u.Value())
	if !json.Valid(ev.Value) {
		ev.Value, _ = json.Marshal(string(u.Value()))
	}
	return ev
}

func (e Emitter) say(ctx context.Context, state string, made time.Time, u jetstream.KeyValueEntry) error {
	ev := e.Event(state, made, u)
	data, err := ev.Encode()
	if err != nil {
		return err
	}
	msg := nats.NewMsg(e.Subject(EventName(state)))
	msg.Header.Set("x-event-id", ev.ID())
	msg.Header.Set("x-source", e.Owner)
	msg.Header.Set("x-time", e.now().Format(time.RFC3339Nano))
	msg.Header.Set("content-type", "application/json")
	msg.Data = data
	return e.Publish(ctx, msg, ev.ID())
}

func (e Emitter) mark(ctx context.Context, state string) (Mark, bool) {
	if e.Marks == nil {
		return Mark{}, false
	}
	got, err := e.Marks.Get(ctx, state)
	if err != nil {
		if !errors.Is(err, jetstream.ErrKeyNotFound) {
			e.logf("views: the mark of %s is not read (%v): said from the bucket's end", state, err)
		}
		return Mark{}, false
	}
	m, err := DecodeMark(got.Value())
	if err != nil {
		e.logf("views: the mark of %s is not read (%v): said from the bucket's end", state, err)
		return Mark{}, false
	}
	return m, true
}

func (e Emitter) writeMark(ctx context.Context, state string, m Mark) {
	if e.Marks == nil {
		return
	}
	m.V, m.At = Version, e.now()
	data, err := json.Marshal(m)
	if err != nil {
		e.logf("views: the mark of %s is not written: %v", state, err)
		return
	}
	if _, err := e.Marks.Put(ctx, state, data); err != nil {
		e.logf("views: the mark of %s at %d is not written (%v): a first view answers the one before", state, m.Revision, err)
	}
}