package events
import (
"context"
"encoding/json"
"errors"
"fmt"
"strconv"
"strings"
"time"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
)
// CounterStream holds monotone tallies — clones, fetches, anything the
// forge counts rather than records.
//
// It is a JetStream counter stream, which is a PN-counter CRDT: the
// server keeps the running total and a per-source slot, so two nodes
// incrementing concurrently converge instead of clobbering, and a retry
// after a timeout cannot double-count. That last property is the reason
// this is not a SELECT/UPDATE in the OLTP — a sync round that is retried
// by a flaky client must not add two.
//
// What it deliberately does NOT do is allocate: a counter converges on a
// TOTAL, never on "which number is mine". Anything needing a unique
// value wants KV's revision-checked update instead.
const (
CounterStream = "QUARRYCOUNT"
counterSubject = "quarry.count.>"
incrHeader = "Nats-Incr"
)
// counterConfig. Counter streams are Limits-retention only, and the
// server refuses to turn AllowMsgCounter off once on, so this shape is
// permanent for the stream's life. MaxMsgsPerSubject 1 keeps only the
// running total per counter: the history is the event stream's job
// (RFC 0005), and a tally that kept every increment would grow without
// bound to say a number it already knows.
//
// Declared against the newer jetstream API because AllowMsgCounter
// exists only there; everything else in this package speaks the legacy
// JetStreamContext, and both ride the same connection.
var counterConfig = jetstream.StreamConfig{
Name: CounterStream,
Subjects: []string{counterSubject},
Retention: jetstream.LimitsPolicy,
AllowMsgCounter: true,
AllowDirect: true,
MaxMsgsPerSubject: 1,
}
// CounterSubject is where a repo's tally of one kind lives.
func CounterSubject(repo, name string) string {
return "quarry.count." + repo + "." + name
}
// Count adds delta to a tally.
//
// Best-effort like Raise: a lost increment costs one off a statistic,
// where failing the operation that triggered it would cost a clone. The
// new total is deliberately not returned — reading it back is Total's
// job, and a caller that only increments should not be made to care.
func (s *Server) Count(repo, name string, delta int64) error {
if s == nil {
return nil
}
if repo == "" || name == "" {
return fmt.Errorf("events: a counter needs a repo and a name")
}
m := nats.NewMsg(CounterSubject(repo, name))
// The sign is required by the server's own regex: +1, never 1.
m.Header.Set(incrHeader, fmt.Sprintf("%+d", delta))
if _, err := s.js.PublishMsg(m); err != nil {
return fmt.Errorf("events: count %s/%s: %w", repo, name, err)
}
return nil
}
// Total reads one tally. A counter that has never been incremented is
// zero rather than an error: "nobody has cloned this yet" is an answer,
// not a failure.
func (s *Server) Total(repo, name string) (int64, error) {
if s == nil {
return 0, nil
}
msg, err := s.js.GetLastMsg(CounterStream, CounterSubject(repo, name))
if err != nil {
if errors.Is(err, nats.ErrMsgNotFound) || errors.Is(err, nats.ErrStreamNotFound) {
return 0, nil
}
return 0, fmt.Errorf("events: total %s/%s: %w", repo, name, err)
}
return parseCounter(msg.Data)
}
// parseCounter reads the server's body form, {"val":"<total>"}. The
// total is a STRING in the JSON because the server uses big.Int and a
// JSON number would lose precision past 2^53 — so it is parsed as one
// rather than as a number that happens to look right for a while.
func parseCounter(body []byte) (int64, error) {
var v struct {
Val string `json:"val"`
}
if err := json.Unmarshal(body, &v); err != nil {
return 0, fmt.Errorf("events: counter body %q: %w", body, err)
}
if v.Val == "" {
return 0, nil
}
n, err := strconv.ParseInt(v.Val, 10, 64)
if err != nil {
return 0, fmt.Errorf("events: counter value %q: %w", v.Val, err)
}
return n, nil
}
// addCounterStream creates the tally stream through the newer API. An
// existing stream is left exactly as it is: AllowMsgCounter cannot be
// turned off once on, so a config update here could only ever fail or
// be a no-op, and failing would take down a server whose counters were
// working fine.
func addCounterStream(nc *nats.Conn) error {
js, err := jetstream.New(nc)
if err != nil {
return fmt.Errorf("events: jetstream: %w", err)
}
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
if _, err := js.CreateStream(ctx, counterConfig); err != nil {
if errors.Is(err, jetstream.ErrStreamNameAlreadyInUse) {
return nil
}
return fmt.Errorf("events: counter stream: %w", err)
}
return nil
}
// CounterToken makes one subject token out of an arbitrary name.
//
// Counter subjects gained structure the moment they started carrying
// who and what, and NATS subject grammar is not a place to be relaxed:
// a dot would silently graft a new level onto the tree, and a > or *
// would turn a stored subject into a wildcard that matches its
// siblings. Anything outside the safe set becomes an underscore, and an
// empty name becomes "_" rather than collapsing two dots into one.
//
// The mapping is deliberately lossy and NOT reversible. A display that
// needs the real name must carry it separately; this exists so that a
// hostile account name cannot reshape the counter tree.
func CounterToken(s string) string {
if s == "" {
return "_"
}
out := make([]rune, 0, len(s))
for _, r := range s {
switch {
case r >= 'a' && r <= 'z', r >= 'A' && r <= 'Z', r >= '0' && r <= '9',
r == '-', r == '_':
out = append(out, r)
default:
out = append(out, '_')
}
}
return string(out)
}
// Totals reads every tally under one family — say "clone.client" — and
// returns them keyed by what follows that prefix.
//
// Two round trips per family rather than one: the stream says which
// subjects exist, then each one's running total is read. That is the
// cost of counters being one subject each, which is also what makes
// them converge across nodes. The families here have a handful of
// members (clients in use, accounts that cloned), so the cost is a
// handful of reads on a page an admin opened deliberately.
func (s *Server) Totals(repo, prefix string) (map[string]int64, error) {
if s == nil {
return nil, nil
}
if repo == "" || prefix == "" {
return nil, fmt.Errorf("events: a counter family needs a repo and a prefix")
}
head := CounterSubject(repo, prefix)
info, err := s.js.StreamInfo(CounterStream, &nats.StreamInfoRequest{SubjectsFilter: head + ".>"})
if err != nil {
if errors.Is(err, nats.ErrStreamNotFound) {
return nil, nil
}
return nil, fmt.Errorf("events: totals %s/%s: %w", repo, prefix, err)
}
out := map[string]int64{}
for subject := range info.State.Subjects {
key := strings.TrimPrefix(subject, head+".")
if key == subject || key == "" {
continue
}
msg, err := s.js.GetLastMsg(CounterStream, subject)
if err != nil {
if errors.Is(err, nats.ErrMsgNotFound) {
continue
}
return nil, fmt.Errorf("events: totals %s: %w", subject, err)
}
n, err := parseCounter(msg.Data)
if err != nil {
return nil, err
}
out[key] = n
}
return out, nil
}