package bus

import (
	"encoding/json"
	"errors"
	"fmt"
	"regexp"
	"sort"
	"strconv"
	"strings"
	"time"

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

// A module's log on the bus (novox/hq ADR 0297): a stream of operations, one subject per key, that the
// controller creates from what the module declared under `logs` and issues to the module's own instances
// in its membership. It mirrors a bucket (ADR 0201) in how it is reached: what a module may reach is
// decided here, from its membership, and refused with the reason before anything is sent.
//
// **The runtime uses no consumer of a log.** The bus grants it, per issued log, only the publish of an
// entry, the stream's info and the direct get: an append is a publish, a read is a batch direct get
// (NATS 2.11), the last entry of a key is a direct get by its subject.

// LogIssued is one log an assignment may reach (ADR 0297): its owner's alone, so there are no reads.
type LogIssued struct {
	Name    string `json:"name"`
	Stream  string `json:"stream"`
	Subject string `json:"subject"`
	Writes  bool   `json:"writes,omitempty"`
}

// LogEntry is one operation in a log: its sequence is its identity.
type LogEntry struct {
	Sequence uint64          `json:"sequence"`
	Key      string          `json:"key"`
	Value    json.RawMessage `json:"value"`
	At       time.Time       `json:"at"`
}

// LogPage is one page of a read: Next is the sequence to continue from, 0 when nothing follows; Last
// is the stream's last sequence.
type LogPage struct {
	Entries []LogEntry `json:"entries"`
	Next    uint64     `json:"next"`
	Last    uint64     `json:"last"`
}

// LogInfo is how full a log is.
type LogInfo struct {
	Entries  uint64 `json:"entries"`
	Bytes    uint64 `json:"bytes"`
	MaxBytes uint64 `json:"max_bytes"`
	First    uint64 `json:"first"`
	Last     uint64 `json:"last"`
}

const (
	// LogValueMax is the most one entry's value may hold (ADR 0297 §3); the stream's MaxMsgSize is 4 KiB
	// more, for the headers an append carries.
	LogValueMax = 256 * 1024
	// LogKeyMax is the longest key.
	LogKeyMax = 255
	// LogReadDefault and LogReadMax are a page's entries when none is asked, and at most.
	LogReadDefault = 100
	LogReadMax     = 1000
	// LogPageBytes bounds the bytes of one page.
	LogPageBytes = 4 * 1024 * 1024
)

// logToken is one name of a key: letters, digits, `_` and `-`.
var logToken = regexp.MustCompile(`^[A-Za-z0-9_-]+$`)

func checkLogKey(key string) error {
	if key == "" || len(key) > LogKeyMax {
		return fmt.Errorf("%q is not a log key: one or more names of letters, digits, _ and -, joined by "+
			"dots, at most %d bytes (novox/hq ADR 0297)", key, LogKeyMax)
	}
	for _, t := range strings.Split(key, ".") {
		if !logToken.MatchString(t) {
			return fmt.Errorf("%q is not a log key: one or more names of letters, digits, _ and -, joined by "+
				"dots, at most %d bytes (novox/hq ADR 0297)", key, LogKeyMax)
		}
	}
	return nil
}

// logFilter is a read's key pattern as a subject filter under the log's subject: `*` for one name, `>`
// for the rest; empty is every key.
func logFilter(l LogIssued, pattern string) (string, error) {
	if pattern == "" {
		return l.Subject + ".>", nil
	}
	parts := strings.Split(pattern, ".")
	for i, p := range parts {
		switch {
		case p == ">" && i == len(parts)-1, p == "*", logToken.MatchString(p):
		default:
			return "", fmt.Errorf("%q is not a key pattern: names, `*` for one and `>` for the rest, last "+
				"(novox/hq ADR 0297)", pattern)
		}
	}
	if len(pattern) > LogKeyMax {
		return "", fmt.Errorf("%q is longer than a key can be (%d bytes)", pattern, LogKeyMax)
	}
	return l.Subject + "." + pattern, nil
}

// issuedLog is the log a module may reach by a name, or the reason it may not.
func (c *Conn) issuedLog(module, name string) (LogIssued, error) {
	m := c.Membership(module)
	if m == nil {
		return LogIssued{}, fmt.Errorf("%s has no membership issued on %s yet, so no log of it is reachable "+
			"until the mesh issues one (novox/hq ADR 0297)", module, c.node)
	}
	var names []string
	for _, l := range m.Logs {
		if l.Name == name {
			return l, nil
		}
		names = append(names, l.Name)
	}
	sort.Strings(names)
	issued := "none"
	if len(names) > 0 {
		issued = strings.Join(names, ", ")
	}
	return LogIssued{}, fmt.Errorf("%s keeps no log called %q: a module declares the logs it keeps under `logs`, "+
		"and a log is its owner's alone; it was issued: %s (novox/hq ADR 0297)", module, name, issued)
}

// notOnTheBus says a log is issued and its stream is not there yet.
func notOnTheBus(l LogIssued) error {
	return fmt.Errorf("the log %q is issued and its stream %s is not on the bus yet: the controller creates "+
		"it from the catalogue on its next raise (novox/hq ADR 0297)", l.Name, l.Stream)
}

// LogAppend writes one entry for a key, as the module, and answers its sequence. With expect, it is
// written only when the key's last entry is still at that sequence (0: the key has none); otherwise it
// is refused as a lost compare-and-set, the same error a lost state update answers (ADR 0259).
func (c *Conn) LogAppend(module, name, key string, value json.RawMessage, expect *uint64) (uint64, error) {
	l, err := c.issuedLog(module, name)
	if err != nil {
		return 0, err
	}
	if !l.Writes {
		return 0, fmt.Errorf("%s was issued the log %s without writing it (novox/hq ADR 0297)", module, name)
	}
	if err := checkLogKey(key); err != nil {
		return 0, err
	}
	if len(value) == 0 || !json.Valid(value) {
		return 0, fmt.Errorf("a log entry's value is JSON (novox/hq ADR 0297)")
	}
	if len(value) > LogValueMax {
		return 0, fmt.Errorf("a log entry's value is at most %d bytes (256 KiB), and this one is %d (novox/hq ADR 0297)",
			LogValueMax, len(value))
	}
	if field := credentialField(value); field != "" {
		return 0, fmt.Errorf("%s's %s.%s carries a field %q, which names a credential: no secret is kept in a log, "+
			"which keeps it for good. Name the secret and fetch it on request/reply (novox/hq ADR 0201 §13, ADR 0297)",
			module, name, key, field)
	}
	msg := nats.NewMsg(l.Subject + "." + key)
	msg.Data = value
	if expect != nil {
		msg.Header.Set(jetstream.ExpectedLastSubjSeqHeader, strconv.FormatUint(*expect, 10))
	}
	reply, err := c.nc.RequestMsg(msg, StateTimeout)
	if errors.Is(err, nats.ErrNoResponders) {
		return 0, notOnTheBus(l)
	}
	if err != nil {
		return 0, err
	}
	var ack struct {
		Seq   uint64 `json:"seq"`
		Error *struct {
			Code        int    `json:"code"`
			ErrCode     uint16 `json:"err_code"`
			Description string `json:"description"`
		} `json:"error"`
	}
	if err := json.Unmarshal(reply.Data, &ack); err != nil {
		return 0, fmt.Errorf("the bus answered the append to %s with what is not an acknowledgement: %q", name, reply.Data)
	}
	if ack.Error != nil {
		// The key's last entry is no longer the one expected: a lost compare-and-set, answered as a lost
		// state update is (10164 is the same refusal from a replicated stream).
		if code := jetstream.ErrorCode(ack.Error.ErrCode); code == jetstream.JSErrCodeStreamWrongLastSequence ||
			code == jetstream.JSErrCodeStreamWrongLastSequenceConstant {
			return 0, changed{name + "." + key}
		}
		return 0, fmt.Errorf("the bus refused the append to %s.%s: %s (%d)", name, key, ack.Error.Description, ack.Error.ErrCode)
	}
	if ack.Seq == 0 {
		return 0, fmt.Errorf("the bus answered the append to %s.%s with no sequence", name, key)
	}
	return ack.Seq, nil
}

// LogRead is the entries from a sequence on, of the keys matching a pattern, at most limit of them and
// about 4 MiB, by one batch direct get.
func (c *Conn) LogRead(module, name string, from uint64, keys string, limit int) (*LogPage, error) {
	l, err := c.issuedLog(module, name)
	if err != nil {
		return nil, err
	}
	switch {
	case limit == 0:
		limit = LogReadDefault
	case limit < 0 || limit > LogReadMax:
		return nil, fmt.Errorf("a read's limit is 1 to %d entries, %d when none is given, not %d (novox/hq ADR 0297)",
			LogReadMax, LogReadDefault, limit)
	}
	filter, err := logFilter(l, keys)
	if err != nil {
		return nil, err
	}
	if from == 0 {
		from = 1
	}
	req, _ := json.Marshal(map[string]any{"seq": from, "next_by_subj": filter, "batch": limit, "max_bytes": LogPageBytes})
	inbox := c.nc.NewInbox()
	sub, err := c.nc.SubscribeSync(inbox)
	if err != nil {
		return nil, err
	}
	defer func() { _ = sub.Unsubscribe() }()
	if err := c.nc.PublishRequest("$JS.API.DIRECT.GET."+l.Stream, inbox, req); err != nil {
		return nil, err
	}
	page := &LogPage{Entries: []LogEntry{}}
	deadline := time.Now().Add(StateTimeout)
	for done := false; !done; {
		msg, err := sub.NextMsg(time.Until(deadline))
		if err != nil {
			if errors.Is(err, nats.ErrNoResponders) {
				return nil, notOnTheBus(l)
			}
			if errors.Is(err, nats.ErrTimeout) {
				return nil, fmt.Errorf("the read of %s did not end in %s", name, StateTimeout)
			}
			return nil, err
		}
		switch status := msg.Header.Get("Status"); status {
		case "":
			e, err := logEntryOf(l, msg)
			if err != nil {
				return nil, err
			}
			page.Entries = append(page.Entries, *e)
		case "204": // the end of the batch: what is still pending after the last entry sent
			pending, _ := strconv.ParseUint(msg.Header.Get("Nats-Num-Pending"), 10, 64)
			if pending > 0 && len(page.Entries) > 0 {
				page.Next = page.Entries[len(page.Entries)-1].Sequence + 1
			}
			done = true
		case "404": // nothing from that sequence on
			done = true
		case "503":
			return nil, notOnTheBus(l)
		default:
			return nil, fmt.Errorf("the read of %s was refused by the bus: %s %s", name, status, msg.Header.Get("Description"))
		}
	}
	info, err := c.logInfo(l)
	if err != nil {
		return nil, err
	}
	page.Last = info.Last
	return page, nil
}

// LogLast is a key's last entry, or nil when it has none.
func (c *Conn) LogLast(module, name, key string) (*LogEntry, error) {
	l, err := c.issuedLog(module, name)
	if err != nil {
		return nil, err
	}
	if err := checkLogKey(key); err != nil {
		return nil, err
	}
	reply, err := c.nc.Request("$JS.API.DIRECT.GET."+l.Stream+"."+l.Subject+"."+key, nil, StateTimeout)
	if errors.Is(err, nats.ErrNoResponders) {
		return nil, notOnTheBus(l)
	}
	if err != nil {
		return nil, err
	}
	switch status := reply.Header.Get("Status"); status {
	case "":
		return logEntryOf(l, reply)
	case "404":
		return nil, nil
	default:
		return nil, fmt.Errorf("the last entry of %s.%s could not be read: %s %s", name, key, status, reply.Header.Get("Description"))
	}
}

// LogInfo is a log's entries, bytes, cap and first and last sequences.
func (c *Conn) LogInfo(module, name string) (*LogInfo, error) {
	l, err := c.issuedLog(module, name)
	if err != nil {
		return nil, err
	}
	return c.logInfo(l)
}

func (c *Conn) logInfo(l LogIssued) (*LogInfo, error) {
	reply, err := c.nc.Request("$JS.API.STREAM.INFO."+l.Stream, nil, StateTimeout)
	if errors.Is(err, nats.ErrNoResponders) {
		return nil, notOnTheBus(l)
	}
	if err != nil {
		return nil, err
	}
	var info struct {
		Config struct {
			MaxBytes int64 `json:"max_bytes"`
		} `json:"config"`
		State struct {
			Msgs     uint64 `json:"messages"`
			Bytes    uint64 `json:"bytes"`
			FirstSeq uint64 `json:"first_seq"`
			LastSeq  uint64 `json:"last_seq"`
		} `json:"state"`
		Error *struct {
			ErrCode     uint16 `json:"err_code"`
			Description string `json:"description"`
		} `json:"error"`
	}
	if err := json.Unmarshal(reply.Data, &info); err != nil {
		return nil, fmt.Errorf("the bus answered the info of %s with what is not one: %q", l.Name, reply.Data)
	}
	if info.Error != nil {
		if info.Error.ErrCode == uint16(jetstream.JSErrCodeStreamNotFound) {
			return nil, notOnTheBus(l)
		}
		return nil, fmt.Errorf("the info of the log %s was refused: %s (%d)", l.Name, info.Error.Description, info.Error.ErrCode)
	}
	var maxBytes uint64
	if info.Config.MaxBytes > 0 {
		maxBytes = uint64(info.Config.MaxBytes)
	}
	return &LogInfo{Entries: info.State.Msgs, Bytes: info.State.Bytes, MaxBytes: maxBytes,
		First: info.State.FirstSeq, Last: info.State.LastSeq}, nil
}

// logEntryOf is an entry a direct get answered: its key is its subject under the log's.
func logEntryOf(l LogIssued, msg *nats.Msg) (*LogEntry, error) {
	subject := msg.Header.Get("Nats-Subject")
	key, ok := strings.CutPrefix(subject, l.Subject+".")
	if !ok {
		return nil, fmt.Errorf("the bus answered an entry of %s on %q, which is not under %s", l.Name, subject, l.Subject)
	}
	seq, err := strconv.ParseUint(msg.Header.Get("Nats-Sequence"), 10, 64)
	if err != nil {
		return nil, fmt.Errorf("the bus answered an entry of %s with no sequence", l.Name)
	}
	at, err := time.Parse(time.RFC3339Nano, msg.Header.Get("Nats-Time-Stamp"))
	if err != nil {
		return nil, fmt.Errorf("the bus answered entry %d of %s with no time: %v", seq, l.Name, err)
	}
	return &LogEntry{Sequence: seq, Key: key, Value: valueOf(msg.Data), At: at.UTC()}, nil
}