package kv

// Append-only event streams, as rows of the kv table.
//
// A stream entry is the row
//
//	ev:<stream>/<ms:13>-<seq:10>  →  JSON object of the entry's string fields
//
// Zero padding makes key order equal ID order, so reading "after an ID" is a
// plain range read and trimming is a range delete. On the wire IDs are the
// unpadded Redis form (`1700000000000-0`): meta-sort, meta-share and
// meta-stremio persist SSE cursors to disk across the Redis → SQLite switch,
// and those compare as <ms>-<seq> pairs.
//
// The last issued ID is kept in `evhwm:<stream>` and updated in the same
// transaction as the append. That high-water mark, not the newest surviving
// row, seeds the next ID — so Clear (the watcher's `reset`) never reissues an
// ID an old cursor would treat as already seen.

import (
	"context"
	"encoding/json"
	"fmt"
	"strconv"
	"strings"
	"time"
)

// Entry is one stream entry.
type Entry struct {
	ID     string
	Values map[string]string
}

type streamID struct{ ms, seq int64 }

func (id streamID) String() string {
	return strconv.FormatInt(id.ms, 10) + "-" + strconv.FormatInt(id.seq, 10)
}

func (id streamID) padded() string { return fmt.Sprintf("%013d-%010d", id.ms, id.seq) }

func (id streamID) less(o streamID) bool {
	return id.ms < o.ms || (id.ms == o.ms && id.seq < o.seq)
}

// ParseID parses a Redis-style stream ID. "", "0", "0-0", "-" all parse as the
// zero ID (before everything). A bare "<ms>" means "<ms>-0".
func ParseID(s string) (ms, seq int64, err error) {
	if s == "" || s == "-" || s == "0" {
		return 0, 0, nil
	}
	msPart, seqPart, hasSeq := strings.Cut(s, "-")
	ms, err = strconv.ParseInt(msPart, 10, 64)
	if err != nil || ms < 0 {
		return 0, 0, fmt.Errorf("invalid stream id %q", s)
	}
	if hasSeq {
		seq, err = strconv.ParseInt(seqPart, 10, 64)
		if err != nil || seq < 0 {
			return 0, 0, fmt.Errorf("invalid stream id %q", s)
		}
	}
	return ms, seq, nil
}

func streamPrefix(stream string) string { return "ev:" + stream + "/" }
func hwmKey(stream string) string       { return "evhwm:" + stream }

// streamState is the in-memory view of a stream's high-water mark plus the
// broadcast channel subscribers wait on. Updated only after a commit.
type streamState struct {
	loaded    bool
	hwm       streamID
	sinceTrim int64 // committed appends since the last trim
	notify    chan struct{}
}

// streamPending is what a transaction did to one stream, applied to
// streamState on commit (and discarded on rollback).
type streamPending struct {
	hwm        streamID
	appends    int64
	trimmed    bool
	lastTrimAt int64 // value of appends when this transaction last trimmed
}

func (s *Store) state(stream string) *streamState {
	st := s.streams[stream]
	if st == nil {
		st = &streamState{notify: make(chan struct{})}
		s.streams[stream] = st
	}
	return st
}

// Wait returns a channel closed by the next commit that appends to stream.
// Take it BEFORE reading, so an append between the read and the wait is not
// missed.
func (s *Store) Wait(stream string) <-chan struct{} {
	s.streamMu.Lock()
	defer s.streamMu.Unlock()
	return s.state(stream).notify
}

func (s *Store) afterCommit(pending map[string]*streamPending) {
	if len(pending) == 0 {
		return
	}
	s.streamMu.Lock()
	defer s.streamMu.Unlock()
	for name, p := range pending {
		st := s.state(name)
		st.loaded = true
		st.hwm = p.hwm
		if p.trimmed {
			st.sinceTrim = p.appends - p.lastTrimAt
		} else {
			st.sinceTrim += p.appends
		}
		close(st.notify)
		st.notify = make(chan struct{})
	}
}

// hwm returns the stream's last issued ID as seen by this transaction.
func (t *Tx) hwm(stream string) (streamID, error) {
	if p := t.g.pending[stream]; p != nil {
		return p.hwm, nil
	}
	t.s.streamMu.Lock()
	st := t.s.state(stream)
	loaded, hwm := st.loaded, st.hwm
	t.s.streamMu.Unlock()
	if loaded {
		return hwm, nil
	}
	v, ok, err := t.Get(t.ctx, hwmKey(stream))
	if err != nil {
		return streamID{}, err
	}
	if !ok {
		return streamID{}, nil
	}
	ms, seq, err := ParseID(v)
	if err != nil {
		return streamID{}, err
	}
	return streamID{ms, seq}, nil
}

// Append adds an entry to stream and returns its ID. The stream is trimmed to
// about maxLen entries (maxLen <= 0: unbounded).
func (t *Tx) Append(stream string, maxLen int64, fields map[string]string) (string, error) {
	last, err := t.hwm(stream)
	if err != nil {
		return "", fmt.Errorf("stream %s hwm: %w", stream, err)
	}
	id := streamID{ms: time.Now().UnixMilli()}
	if !last.less(id) {
		id = streamID{ms: last.ms, seq: last.seq + 1}
	}
	body, err := json.Marshal(fields)
	if err != nil {
		return "", err
	}
	if _, err := t.stmt(t.s.stInsert).ExecContext(t.ctx, streamPrefix(stream)+id.padded(), string(body)); err != nil {
		return "", fmt.Errorf("append %s: %w", stream, err)
	}
	if t.g.pending == nil {
		t.g.pending = make(map[string]*streamPending)
	}
	p := t.g.pending[stream]
	if p == nil {
		p = &streamPending{}
		t.g.pending[stream] = p
	}
	p.hwm = id
	p.appends++
	// Trim in steps, not per append: MAXLEN ~ semantics, like the XADD it
	// replaces. The count spans transactions (most append one event each).
	if maxLen > 0 {
		t.s.streamMu.Lock()
		since := t.s.state(stream).sinceTrim
		t.s.streamMu.Unlock()
		if !p.trimmed {
			since += p.appends
		} else {
			since = p.appends - p.lastTrimAt
		}
		if since >= trimEvery {
			if err := t.trim(stream, maxLen); err != nil {
				return "", err
			}
			p.trimmed, p.lastTrimAt = true, p.appends
		}
	}
	return id.String(), nil
}

// trimEvery is how many appends pass between trims. A trim locates its cut
// point by walking maxLen entries back from the newest, so it is kept rare;
// the cost is overshooting maxLen by at most trimEvery entries.
var trimEvery int64 = 10_000 // var so tests can lower it

func (t *Tx) trim(stream string, maxLen int64) error {
	prefix := streamPrefix(stream)
	var cut string
	err := t.sqlTx().QueryRowContext(t.ctx,
		`SELECT k FROM kv WHERE k >= ? AND k < ? ORDER BY k DESC LIMIT 1 OFFSET ?`,
		prefix, PrefixEnd(prefix), maxLen).Scan(&cut)
	if err != nil {
		return nil // fewer than maxLen entries (sql.ErrNoRows): nothing to trim
	}
	_, err = t.sqlTx().ExecContext(t.ctx, `DELETE FROM kv WHERE k >= ? AND k <= ?`, prefix, cut)
	return err
}

// flushHWM persists each touched stream's high-water mark — once per
// transaction, not per append.
func (t *Tx) flushHWM() error {
	for stream, p := range t.g.pending {
		if _, err := t.stmt(t.s.stPut).ExecContext(t.ctx, hwmKey(stream), p.hwm.String()); err != nil {
			return fmt.Errorf("stream %s hwm: %w", stream, err)
		}
	}
	return nil
}

// TrimAll trims stream to maxLen inside the transaction regardless of how many
// appends this transaction made. Used after an append burst split over many
// short transactions.
func (t *Tx) TrimAll(stream string, maxLen int64) error { return t.trim(stream, maxLen) }

// ClearStream deletes every entry of stream, keeping its high-water mark.
func (t *Tx) ClearStream(stream string) error {
	prefix := streamPrefix(stream)
	_, err := t.sqlTx().ExecContext(t.ctx, `DELETE FROM kv WHERE k >= ? AND k < ?`, prefix, PrefixEnd(prefix))
	return err
}

func decodeEntry(stream, k, v string) (Entry, error) {
	padded := strings.TrimPrefix(k, streamPrefix(stream))
	ms, seq, err := ParseID(padded)
	if err != nil {
		return Entry{}, err
	}
	var values map[string]string
	if err := json.Unmarshal([]byte(v), &values); err != nil {
		return Entry{}, fmt.Errorf("decode %s: %w", k, err)
	}
	return Entry{ID: streamID{ms, seq}.String(), Values: values}, nil
}

func decodeEntries(stream string, rows []KV) ([]Entry, error) {
	out := make([]Entry, 0, len(rows))
	for _, kv := range rows {
		e, err := decodeEntry(stream, kv.K, kv.V)
		if err != nil {
			return nil, err
		}
		out = append(out, e)
	}
	return out, nil
}

// ReadAfter returns up to n entries strictly after id, oldest first. "0",
// "0-0" and "" read from the beginning.
func (s *Store) ReadAfter(ctx context.Context, stream, id string, n int) ([]Entry, error) {
	ms, seq, err := ParseID(id)
	if err != nil {
		return nil, err
	}
	prefix := streamPrefix(stream)
	after := ""
	if ms != 0 || seq != 0 {
		after = prefix + streamID{ms, seq}.padded()
	}
	rows, err := rangeQuery(ctx, s.r, prefix, after, n)
	if err != nil {
		return nil, err
	}
	return decodeEntries(stream, rows)
}

// Oldest returns the oldest surviving entry, or ok=false for an empty stream.
func (s *Store) Oldest(ctx context.Context, stream string) (Entry, bool, error) {
	rows, err := rangeQuery(ctx, s.r, streamPrefix(stream), "", 1)
	if err != nil || len(rows) == 0 {
		return Entry{}, false, err
	}
	e, err := decodeEntry(stream, rows[0].K, rows[0].V)
	return e, err == nil, err
}

// LastID returns the last ID ever issued on stream ("0-0" if none) — the
// cursor meaning "only entries appended from now on".
func (s *Store) LastID(ctx context.Context, stream string) (string, error) {
	s.streamMu.Lock()
	st := s.state(stream)
	loaded, hwm := st.loaded, st.hwm
	s.streamMu.Unlock()
	if loaded {
		return hwm.String(), nil
	}
	v, ok, err := s.Get(ctx, hwmKey(stream))
	if err != nil {
		return "", err
	}
	if !ok {
		return "0-0", nil
	}
	return v, nil
}

// Len is the number of surviving entries.
func (s *Store) Len(ctx context.Context, stream string) (int64, error) {
	return count(ctx, s.r, streamPrefix(stream))
}

// RevRange returns up to n entries newest first, stopping at (excluding)
// entries at or before afterID when afterID is non-empty.
func (s *Store) RevRange(ctx context.Context, stream, afterID string, n int) ([]Entry, error) {
	prefix := streamPrefix(stream)
	query := `SELECT k, v FROM kv WHERE k > ? AND k < ? ORDER BY k DESC`
	lo := prefix
	if afterID != "" {
		ms, seq, err := ParseID(afterID)
		if err != nil {
			return nil, err
		}
		lo = prefix + streamID{ms, seq}.padded()
	}
	args := []any{lo, PrefixEnd(prefix)}
	if n > 0 {
		query += ` LIMIT ?`
		args = append(args, n)
	}
	rows, err := s.r.QueryContext(ctx, query, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var kvs []KV
	for rows.Next() {
		var kv KV
		if err := rows.Scan(&kv.K, &kv.V); err != nil {
			return nil, err
		}
		kvs = append(kvs, kv)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return decodeEntries(stream, kvs)
}