counters/internal/txlog/txlog.go

// Package txlog writes the counting transaction log: one block of lines per
// request, in the shape of the rspamd Lua plugin's log so existing grep
// habits keep working, with rspamd's task id replaced by the trace id:
//
//	2026-10-06 01:12:06.412 <ad8525> out; local domain: example.com account_id=7
//	2026-10-06 01:12:06.412 <ad8525> out; determined limit (level=[A] flow=[o] key=[cnt/A/x@example.com/d/2026-10-06/o]) => 100
//	2026-10-06 01:12:06.412 <ad8525> out; tracked address counters for x@example.com => count=12 limit=100 reached=false
//
// Each block is written with a single Write so concurrent requests never
// interleave. With format "json" the same events become one JSON object per
// line.
package txlog

import (
	"bytes"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"sort"
	"strings"
	"time"

	"migadu_counters/internal/counting"
	"migadu_counters/internal/domains"
	"migadu_counters/internal/notify"
)

// Logger writes transaction blocks to w.
type Logger struct {
	w    io.Writer
	json bool
	now  func() time.Time
}

// New creates a Logger. format is "text" (default) or "json"; now may be nil.
func New(w io.Writer, format string, now func() time.Time) *Logger {
	if now == nil {
		now = time.Now
	}
	return &Logger{w: w, json: format == "json", now: now}
}

// block accumulates the lines of one request.
type block struct {
	l     *Logger
	ts    time.Time
	trace string
	route string
	buf   bytes.Buffer
}

func (l *Logger) begin(trace, route string) *block {
	return &block{l: l, ts: l.now().UTC(), trace: trace, route: route}
}

// line appends one event: msg in text mode, fields in json mode.
func (b *block) line(event, msg string, fields map[string]any) {
	if b.l.json {
		rec := map[string]any{
			"ts":       b.ts.Format("2006-01-02T15:04:05.000Z07:00"),
			"trace_id": b.trace,
			"route":    b.route,
			"event":    event,
		}
		for k, v := range fields {
			rec[k] = v
		}
		data, err := json.Marshal(rec)
		if err != nil {
			data = []byte(fmt.Sprintf(`{"event":"marshal_error","error":%q}`, err.Error()))
		}
		b.buf.Write(data)
		b.buf.WriteByte('\n')
		return
	}
	trace := b.trace
	if trace == "" {
		trace = "-"
	}
	fmt.Fprintf(&b.buf, "%s <%s> %s; %s\n", b.ts.Format("2006-01-02 15:04:05.000"), trace, b.route, msg)
}

func (b *block) flush() { _, _ = b.l.w.Write(b.buf.Bytes()) }

// Outcome logs an executed plan. dispatched[i] tells, for a Results[i] that
// reached its limit, whether a notification was handed to the notifier.
func (l *Logger) Outcome(out *counting.Outcome, dispatched []bool) {
	b := l.begin(out.TraceID, out.Route)
	b.request(out.Plan)
	b.trail(out.Plan)
	if out.Unknown != nil {
		b.line("redis_unknown", fmt.Sprintf("redis outcome unknown: %v, re-read from master", out.Unknown),
			map[string]any{"error": out.Unknown.Error(), "reread": true})
	}
	for _, r := range out.Results {
		if r.Err != nil {
			b.line("redis_key_error", fmt.Sprintf("redis error on key %s: %v (not counted)", r.Key, r.Err),
				map[string]any{"level": r.Level, "id": r.ID, "flow": r.Flow, "key": r.Key, "error": r.Err.Error()})
			continue
		}
		b.tracked(r, out.Unknown != nil)
	}
	for i, r := range out.Results {
		if r.Reached {
			b.reached(r, i < len(dispatched) && dispatched[i])
		}
	}
	if out.Skipped != "" {
		b.line("skipped", "skipped: "+out.Skipped, map[string]any{"reason": out.Skipped})
	}
	b.flush()
}

// Failed logs a plan whose increments could not be written to Redis.
func (l *Logger) Failed(plan *counting.Plan, err error) {
	b := l.begin(plan.TraceID, plan.Route)
	b.request(plan)
	b.trail(plan)
	b.line("redis_error", fmt.Sprintf("redis error: %v (nothing counted)", err), map[string]any{"error": err.Error()})
	b.flush()
}

// Dropped logs a request that was rejected because the counting queue was full.
func (l *Logger) Dropped(route, traceID string) {
	b := l.begin(traceID, route)
	b.line("dropped", "dropped: queue full (not counted)", map[string]any{"reason": "queue full"})
	b.flush()
}

// Error logs a request that failed before a plan existed (domain lookup error).
func (l *Logger) Error(route, traceID string, err error) {
	b := l.begin(traceID, route)
	b.line("error", fmt.Sprintf("error: %v (nothing counted)", err), map[string]any{"error": err.Error()})
	b.flush()
}

// NotificationSent logs a delivered web app notification (notify hook).
func (l *Logger) NotificationSent(ev notify.Event, url string, attempt int) {
	b := l.begin(ev.TraceID, ev.Route)
	b.line("notification_sent", fmt.Sprintf("sent limit exceeded notification to url %q", url),
		map[string]any{"url": url, "attempt": attempt, "level": ev.Level, "id": ev.ID, "flow": ev.Flow})
	b.flush()
}

// NotificationFailed logs a notification given up on (notify hook).
func (l *Logger) NotificationFailed(ev notify.Event, url string, attempts int, err error) {
	b := l.begin(ev.TraceID, ev.Route)
	b.line("notification_failed", fmt.Sprintf("failed limit exceeded notification to url %q after %d attempts: %v", url, attempts, err),
		map[string]any{"url": url, "attempts": attempts, "error": err.Error(), "level": ev.Level, "id": ev.ID, "flow": ev.Flow})
	b.flush()
}

func (b *block) request(p *counting.Plan) {
	if p.Route == counting.RouteOut {
		msg := fmt.Sprintf("request auth_user=%s recipient=%s mail_from=%s", p.AuthUser, p.Recipient, p.MailFrom)
		fields := map[string]any{"auth_user": p.AuthUser, "recipient": p.Recipient, "mail_from": p.MailFrom}
		if p.LoginSuffix != "" {
			msg += " login_suffix=" + p.LoginSuffix
			fields["login_suffix"] = p.LoginSuffix
		}
		b.line("request", msg, fields)
	} else {
		b.line("request", fmt.Sprintf("request mail_from=%s recipient=%s junk=%t", p.MailFrom, p.Recipient, p.Junk),
			map[string]any{"mail_from": p.MailFrom, "recipient": p.Recipient, "junk": p.Junk})
	}
	for _, u := range p.Unusable {
		shown := u.Value
		if shown == "" {
			shown = "<empty>"
		}
		b.line("not_usable", fmt.Sprintf("%s address not usable: %s (%s, not counted)", u.Role, shown, u.Reason),
			map[string]any{"role": u.Role, "value": u.Value, "reason": u.Reason})
	}
}

// trail logs the domain lookups and the limits determined for each.
func (b *block) trail(p *counting.Plan) {
	for _, lk := range p.Lookups {
		if lk.Domain == nil {
			if lk.Err == nil || errors.Is(lk.Err, domains.ErrNotFound) {
				b.line("not_in_db", fmt.Sprintf("%s domain not in domains db: %s (not counted)", lk.Role, lk.DomainName),
					map[string]any{"role": lk.Role, "domain": lk.DomainName})
			} else if errors.Is(lk.Err, domains.ErrMalformed) {
				detail := strings.TrimPrefix(lk.Err.Error(), domains.ErrMalformed.Error()+": ")
				b.line("malformed_row", fmt.Sprintf("%s domain row malformed in domains db: %s: %s (not counted)", lk.Role, lk.DomainName, detail),
					map[string]any{"role": lk.Role, "domain": lk.DomainName, "error": detail})
			} else {
				b.line("lookup_error", fmt.Sprintf("%s domain lookup failed: %s: %v", lk.Role, lk.DomainName, lk.Err),
					map[string]any{"role": lk.Role, "domain": lk.DomainName, "error": lk.Err.Error()})
			}
			continue
		}
		d := lk.Domain
		b.line("local_domain", fmt.Sprintf("local domain: %s account_id=%d", d.Name, d.OrgID),
			map[string]any{"role": lk.Role, "domain": d.Name, "org_id": d.OrgID})
		local := lk.Address
		if i := strings.LastIndex(local, "@"); i >= 0 {
			local = local[:i]
		}
		shown, other := summarize(d.Limits, local)
		msg := fmt.Sprintf("domain: %s limits_map=%s", d.Name, renderLimits(shown, d.Limits))
		if other > 0 {
			msg += fmt.Sprintf(" (+%d other overrides)", other)
		}
		limitsJSON := make(map[string]int64, len(shown))
		for _, k := range shown {
			limitsJSON[k] = d.Limits[k]
		}
		b.line("limits", msg, map[string]any{"domain": d.Name, "limits": limitsJSON, "other_overrides": other})

		flow := domains.FlowIn
		if lk.Role == "sender" {
			flow = domains.FlowOut
		}
		for _, inc := range p.Increments {
			if inc.By != lk.Address || inc.Flow != flow {
				continue
			}
			b.line("determined_limit",
				fmt.Sprintf("determined limit (level=[%s] flow=[%s] key=[%s]) => %s", inc.Level, inc.Flow, inc.Key, limitText(inc)),
				map[string]any{"level": inc.Level, "flow": inc.Flow, "key": inc.Key, "limit": limitJSON(inc), "limited": inc.Limited})
		}
	}
}

func (b *block) tracked(r counting.Result, reread bool) {
	msg := fmt.Sprintf("tracked %s counters for %s => count=%d limit=%s reached=%t", levelWord(r.Level), r.ID, r.Count, limitText(r.Increment), r.Reached)
	fields := map[string]any{"level": r.Level, "id": r.ID, "flow": r.Flow, "key": r.Key, "count": r.Count, "limit": limitJSON(r.Increment), "reached": r.Reached}
	if reread {
		msg += " (re-read)"
		fields["reread"] = true
	}
	b.line("tracked", msg, fields)
}

func (b *block) reached(r counting.Result, dispatched bool) {
	name := notify.URLName(r.Level)
	verdict := "notifying " + name
	if !dispatched {
		verdict = "no " + name + " configured, not notifying"
	}
	b.line("limit_reached",
		fmt.Sprintf("limit reached (level=[%s] id=[%s] flow=[%s] by=[%s]) => %s", r.Level, r.ID, r.Flow, r.By, verdict),
		map[string]any{"level": r.Level, "id": r.ID, "flow": r.Flow, "by": r.By, "count": r.Count, "limit": r.Limit, "notifying": dispatched, "url_name": name})
}

// summarize picks the limit keys worth printing for one address: the base
// keys and the address's own overrides. Some domains carry thousands of
// per-mailbox overrides (boxmailhere.com: 9 798); dumping them all would make
// a 190 KB log line.
func summarize(l domains.Limits, local string) (shown []string, other int) {
	keys := make([]string, 0, len(l))
	for k := range l {
		keys = append(keys, k)
	}
	sort.Strings(keys)
	for _, k := range keys {
		if !strings.Contains(k, "@/") || strings.HasPrefix(k, local+"@/") {
			shown = append(shown, k)
		} else {
			other++
		}
	}
	return shown, other
}

func renderLimits(keys []string, l domains.Limits) string {
	parts := make([]string, len(keys))
	for i, k := range keys {
		parts[i] = fmt.Sprintf("%s=%d", k, l[k])
	}
	return "{" + strings.Join(parts, ", ") + "}"
}

func levelWord(level string) string {
	switch level {
	case domains.AddressLevel:
		return "address"
	case domains.DomainLevel:
		return "domain"
	case domains.OrganizationLevel:
		return "account" // the Lua plugin's word for the organisation level
	}
	return level
}

func limitText(inc counting.Increment) string {
	if !inc.Limited {
		return "unlimited"
	}
	return fmt.Sprintf("%d", inc.Limit)
}

func limitJSON(inc counting.Increment) any {
	if !inc.Limited {
		return nil
	}
	return inc.Limit
}