aml/pkg/api/monitor_test.go

package api

import (
	"context"
	"encoding/json"
	"net/http"
	"os"
	"strings"
	"testing"
	"time"

	"github.com/luxfi/aml/pkg/dictionary"
	"github.com/luxfi/aml/pkg/engine"
	"github.com/luxfi/aml/pkg/lists"
	"github.com/luxfi/aml/pkg/reference"
	"github.com/luxfi/aml/pkg/suppress"
	"github.com/luxfi/aml/pkg/types"
	"github.com/luxfi/aml/pkg/watch"
)

// The ingest path meets the five planes here, and this is where that meeting is
// checked end to end: a rule fires, an activation is written, a declared
// suppression changes the answer the caller gets, and the field catalog sees the
// payload.
//
// The planes' own tests establish that each one is correct. This one establishes
// that ingest is WIRED to them — the failure it exists to catch is five correct
// planes that nothing calls.

// monitored is the deployment. It is an alias of plane, kept because the tests
// that use it are about ingest meeting the planes rather than about ingest
// alone — and there is one assembly, so there is nothing for it to add.
func monitored(t *testing.T, rules ...types.Rule) *Handler {
	t.Helper()
	return plane(t, rules...)
}

// payment is the wire shape the ingest route takes: the transaction at the top
// level, with the customer beside it.
func payment(id string, usd float64) map[string]any {
	return map[string]any{
		"id": id, "user_id": "u1", "account_id": "acct-1",
		"currency": "USD", "notional": usd, "timestamp": time.Now().UTC(),
		"entity": map[string]any{"id": "u1", "name": "Acme Trading"},
	}
}

// TestIngestRecordsWhatTheRulesDid.
func TestIngestRecordsWhatTheRulesDid(t *testing.T) {
	h := monitored(t)
	ctx := context.Background()

	rec := ingest(t, h, payment("tx-1", 25_000))
	if rec.Code != http.StatusOK {
		t.Fatalf("ingest = %d: %s", rec.Code, rec.Body.String())
	}
	var out types.EvalResult
	if err := json.Unmarshal(rec.Body.Bytes(), &out); err != nil {
		t.Fatal(err)
	}
	if out.Action != types.ActionReport {
		t.Fatalf("the rule asked for report, got %q", out.Action)
	}

	feed, err := h.Planes.Watch.Feed(ctx, acme, &watch.FeedIn{})
	if err != nil {
		t.Fatal(err)
	}
	if len(feed.Activations) != 1 {
		t.Fatalf("the rule fired and nothing recorded it: %+v", feed.Activations)
	}
	a := feed.Activations[0]
	if a.Rule != "ctr" || a.Tx != "tx-1" || a.Subject.Value != "acct-1" || a.Subject.Kind != "account" {
		t.Fatalf("the activation does not describe what happened: %+v", a)
	}
	if a.Action != types.ActionReport || a.Response != types.ActionReport {
		t.Fatalf("with nothing declared, the response is what the rule asked for: %+v", a)
	}

	// And the payload reached the catalog.
	cat, err := h.Planes.Dictionary.Catalog(ctx, acme, &dictionary.CatalogIn{})
	if err != nil {
		t.Fatal(err)
	}
	if cat.Payloads != 1 {
		t.Fatalf("the field catalog did not see the payload: %d", cat.Payloads)
	}
}

// TestADeclaredSuppressionChangesTheAnswer is the whole of what suppression does,
// exercised where a customer would feel it.
func TestADeclaredSuppressionChangesTheAnswer(t *testing.T) {
	h := monitored(t)
	ctx := context.Background()
	if _, err := h.Planes.Suppress.Suppress(ctx, acme, &suppress.SuppressIn{
		Rule: "ctr", Kind: "account", Value: "acct-1",
		Reason: "treasury sweep, agreed with the MLRO 2026-02", By: "a.mensah",
	}); err != nil {
		t.Fatal(err)
	}

	rec := ingest(t, h, payment("tx-2", 25_000))
	if rec.Code != http.StatusOK {
		t.Fatalf("ingest = %d: %s", rec.Code, rec.Body.String())
	}
	var out types.EvalResult
	if err := json.Unmarshal(rec.Body.Bytes(), &out); err != nil {
		t.Fatal(err)
	}
	if out.Action != types.ActionAllow {
		t.Fatalf("a covered detection must not reach the caller as a report: %q", out.Action)
	}
	// The alert itself is still evidence and is still there.
	if len(out.AlertIDs) != 1 {
		t.Fatalf("suppression must not destroy the alert: %+v", out.AlertIDs)
	}

	feed, err := h.Planes.Watch.Feed(ctx, acme, &watch.FeedIn{})
	if err != nil {
		t.Fatal(err)
	}
	if len(feed.Activations) != 1 {
		t.Fatalf("a suppressed detection is recorded, not dropped: %+v", feed.Activations)
	}
	a := feed.Activations[0]
	if !a.Suppressed || a.Cause != watch.CauseSuppressed || a.By == "" {
		t.Fatalf("the activation must name the suppression that covered it: %+v", a)
	}
	if a.Action != types.ActionReport {
		t.Fatalf("what the rule asked for is still on the row: %+v", a)
	}
}

// TestADeclaredRungRaisesTheAnswer.
func TestADeclaredRungRaisesTheAnswer(t *testing.T) {
	h := monitored(t)
	ctx := context.Background()
	if _, err := h.Planes.Watch.Declare(ctx, acme, &watch.DeclareIn{
		Rule: "ctr", Kind: "account", Count: 2, Within: watch.Span(time.Hour), To: types.ActionBlock,
		Reason: "two reportable transactions in an hour on one account", By: "a.mensah",
	}); err != nil {
		t.Fatal(err)
	}

	if rec := ingest(t, h, payment("tx-3", 25_000)); rec.Code != http.StatusOK {
		t.Fatalf("ingest = %d: %s", rec.Code, rec.Body.String())
	}
	rec := ingest(t, h, payment("tx-4", 25_000))
	if rec.Code != http.StatusOK {
		t.Fatalf("ingest = %d: %s", rec.Code, rec.Body.String())
	}
	var out types.EvalResult
	if err := json.Unmarshal(rec.Body.Bytes(), &out); err != nil {
		t.Fatal(err)
	}
	if out.Action != types.ActionBlock {
		t.Fatalf("the declared escalation did not reach the caller: %q", out.Action)
	}
}

// TestARuleReadsTheTenantsOwnDenyList — the list plane reaching the rule
// vocabulary, through the engine's own evaluator.
func TestARuleReadsTheTenantsOwnDenyList(t *testing.T) {
	h := monitored(t)
	ctx := context.Background()
	if _, err := h.Planes.Lists.Declare(ctx, acme, &lists.DeclareIn{
		Name: "ip-deny", Kind: lists.Deny, Class: lists.IP, By: "a.mensah",
	}); err != nil {
		t.Fatal(err)
	}
	if _, err := h.Planes.Lists.Add(ctx, acme, &lists.AddIn{
		Name: "ip-deny", By: "a.mensah",
		Values: []lists.Value{{Value: "198.51.100.7", Reason: "credential stuffing source"}},
	}); err != nil {
		t.Fatal(err)
	}

	// A rule over the list, admitted through the engine's own vocabulary check —
	// which is what refuses the rule outright if the provider is not wired.
	h.Engine = engine.New(engine.Providers{
		Rate:  reference.Rates{USDPer: map[string]float64{}},
		Lists: h.Planes.Lists,
		Zone:  time.UTC,
	})
	if err := h.Engine.SetRules([]types.Rule{{
		ID: "denied-address", Name: "Address on the deny list",
		DSL:      `Tx.IPAddress != "" && Listed("ip-deny", Tx.IPAddress)`,
		Severity: types.SeverityHigh, Weight: 0.4, Action: types.ActionBlock, Enabled: true,
	}}); err != nil {
		t.Fatalf("the deny-list rule must install where the list plane is wired: %v", err)
	}

	body := payment("tx-5", 100)
	body["ip_address"] = "198.51.100.7"
	rec := ingest(t, h, body)
	if rec.Code != http.StatusOK {
		t.Fatalf("ingest = %d: %s", rec.Code, rec.Body.String())
	}
	var out types.EvalResult
	if err := json.Unmarshal(rec.Body.Bytes(), &out); err != nil {
		t.Fatal(err)
	}
	if out.Action != types.ActionBlock {
		t.Fatalf("a payment from a listed address must be blocked, got %q", out.Action)
	}

	// And an address nobody listed passes.
	clean := payment("tx-6", 100)
	clean["ip_address"] = "203.0.113.9"
	rec = ingest(t, h, clean)
	if err := json.Unmarshal(rec.Body.Bytes(), &out); err != nil {
		t.Fatal(err)
	}
	if out.Action != types.ActionAllow {
		t.Fatalf("an address nobody listed must pass, got %q", out.Action)
	}
}

// TestARetriedTransactionIsCountedOnce.
//
// Ingest records a transaction, then its alerts, then their activations. Anything
// after the first write that fails answers 503, and a client that retries a 503
// sends the whole transaction again — so every plane that ingest writes has to be
// able to see the same transaction twice and record it once. The activation plane
// is the one this track added, and a duplicated activation is worse than a
// duplicated row: it is counted in the streak, so a declared repetition policy
// fires on a repeat that never happened.
func TestARetriedTransactionIsCountedOnce(t *testing.T) {
	h := monitored(t)
	ctx := context.Background()

	body := payment("tx-1", 25_000)
	for range 3 {
		if rec := ingest(t, h, body); rec.Code != http.StatusOK {
			t.Fatalf("ingest = %d: %s", rec.Code, rec.Body.String())
		}
	}

	feed, err := h.Planes.Watch.Feed(ctx, acme, &watch.FeedIn{})
	if err != nil {
		t.Fatal(err)
	}
	if len(feed.Activations) != 1 {
		t.Fatalf("three offers of one transaction wrote %d activations: %+v", len(feed.Activations), feed.Activations)
	}
	rates, err := h.Planes.Watch.Rates(ctx, acme, &watch.RatesIn{})
	if err != nil {
		t.Fatal(err)
	}
	if len(rates.Rules) != 1 || rates.Rules[0].Fired != 1 {
		t.Fatalf("the rates counted the retries: %+v", rates.Rules)
	}

	// And a different transaction is still a different activation, so the fix is
	// not "record nothing twice".
	if rec := ingest(t, h, payment("tx-2", 25_000)); rec.Code != http.StatusOK {
		t.Fatalf("ingest = %d: %s", rec.Code, rec.Body.String())
	}
	feed, err = h.Planes.Watch.Feed(ctx, acme, &watch.FeedIn{})
	if err != nil {
		t.Fatal(err)
	}
	if len(feed.Activations) != 2 {
		t.Fatalf("a second transaction did not produce a second activation: %+v", feed.Activations)
	}
}

// TestARetriedTransactionDoesNotEscalateItself.
//
// The sharpest form: a rung raises the response on the second firing, so a
// duplicated activation makes the RETRY of a single transaction escalate itself —
// a payment blocked because the client retried a 503.
func TestARetriedTransactionDoesNotEscalateItself(t *testing.T) {
	h := monitored(t)
	if _, err := h.Planes.Watch.Declare(context.Background(), acme, &watch.DeclareIn{
		Rule: "ctr", Kind: "account", Count: 2, Within: watch.Span(time.Hour), To: types.ActionBlock,
		Reason: "two reportable transactions in an hour on one account", By: "a.mensah",
	}); err != nil {
		t.Fatal(err)
	}

	body := payment("tx-1", 25_000)
	var out types.EvalResult
	for range 2 {
		rec := ingest(t, h, body)
		if rec.Code != http.StatusOK {
			t.Fatalf("ingest = %d: %s", rec.Code, rec.Body.String())
		}
		if err := json.Unmarshal(rec.Body.Bytes(), &out); err != nil {
			t.Fatal(err)
		}
	}
	if out.Action == types.ActionBlock {
		t.Fatal("retrying one transaction escalated it against itself")
	}

	// A genuinely second transaction still escalates, so the rung still works.
	rec := ingest(t, h, payment("tx-2", 25_000))
	if err := json.Unmarshal(rec.Body.Bytes(), &out); err != nil {
		t.Fatal(err)
	}
	if out.Action != types.ActionBlock {
		t.Fatalf("the declared escalation stopped working: %q", out.Action)
	}
}

// TestOneTenantsSuppressionsAreNeverAnothersIngest.
//
// The isolation invariant, at the seam this track changed. A tenant that has
// crowded one rule with suppressions degrades ITS OWN cover check and nothing
// else: another institution's ingest is unaffected, its cover check is complete,
// and neither reads the other's rows.
func TestOneTenantsSuppressionsAreNeverAnothersIngest(t *testing.T) {
	h := monitored(t)
	ctx := context.Background()
	const rival = "hanzo/rival"

	for i := range 6 {
		if _, err := h.Planes.Suppress.Suppress(ctx, acme, &suppress.SuppressIn{
			Rule: "ctr", Kind: "account", Value: "acct-" + string(rune('a'+i)),
			Reason: "reviewed and agreed", By: "a.mensah",
		}); err != nil {
			t.Fatal(err)
		}
	}

	// The other institution's own suppression still covers its own activation, and
	// this institution's six do not appear anywhere in its answer.
	if _, err := h.Planes.Suppress.Suppress(ctx, rival, &suppress.SuppressIn{
		Rule: "ctr", Kind: "account", Value: "acct-1", Reason: "their own decision", By: "b.tran",
	}); err != nil {
		t.Fatal(err)
	}
	cover, err := h.Planes.Suppress.Cover(ctx, rival, &suppress.CoverIn{Rule: "ctr", Kind: "account", Value: "acct-1"})
	if err != nil {
		t.Fatalf("the other tenant's cover check failed: %v", err)
	}
	if !cover.Covered || cover.Partial {
		t.Fatalf("the other tenant's answer was affected by this one's volume: %+v", cover)
	}
	ledger, err := h.Planes.Suppress.Ledger(ctx, rival, &suppress.LedgerIn{})
	if err != nil {
		t.Fatal(err)
	}
	if len(ledger.Suppressions) != 1 {
		t.Fatalf("the other tenant reads %d suppressions, want only its own", len(ledger.Suppressions))
	}

	// And this institution's ingest still works, which is the self-DoS half.
	if rec := ingest(t, h, payment("tx-1", 25_000)); rec.Code != http.StatusOK {
		t.Fatalf("ingest = %d: %s", rec.Code, rec.Body.String())
	}
	feed, err := h.Planes.Watch.Feed(ctx, rival, &watch.FeedIn{})
	if err != nil {
		t.Fatal(err)
	}
	if len(feed.Activations) != 0 {
		t.Fatalf("one tenant's ingest wrote into another's activation plane: %+v", feed.Activations)
	}
}

// TestARealConflictIsAConflictAndNotAnOutage.
//
// The other half of making a retry work: an id reused for a DIFFERENT
// transaction is refused, and the refusal says which kind it is. 503 means "the
// engine could not, try again" and a client that retries a permanent refusal
// retries forever — which is the loop that turns one caller's mistake into a
// queue of transactions that never clears.
func TestARealConflictIsAConflictAndNotAnOutage(t *testing.T) {
	h := monitored(t)

	first := payment("tx-1", 25_000)
	if rec := ingest(t, h, first); rec.Code != http.StatusOK {
		t.Fatalf("ingest = %d: %s", rec.Code, rec.Body.String())
	}
	// Same id, a different amount. That is a different fact, and the ledger has
	// one under this name already.
	second := payment("tx-1", 90_000)
	second["timestamp"] = first["timestamp"]
	rec := ingest(t, h, second)
	if rec.Code != http.StatusConflict {
		t.Fatalf("a reused id with a different transaction = %d: %s, want 409", rec.Code, rec.Body.String())
	}

	// And the first record stands: a conflict refuses the second submission, it
	// does not overwrite what was retained.
	feed, err := h.Planes.Watch.Feed(context.Background(), acme, &watch.FeedIn{})
	if err != nil {
		t.Fatal(err)
	}
	if len(feed.Activations) != 1 {
		t.Fatalf("the conflicting submission left %d activations: %+v", len(feed.Activations), feed.Activations)
	}
}

// TestIngestStampsNoClockOfItsOwn reads the source. A reception clock written
// onto the transaction is what made the retry above impossible to recognise, and
// it is one line to reintroduce.
func TestIngestStampsNoClockOfItsOwn(t *testing.T) {
	raw, err := os.ReadFile("routes.go")
	if err != nil {
		t.Fatal(err)
	}
	for _, banned := range []string{"tx.CreatedAt =", "tx.UpdatedAt ="} {
		if strings.Contains(string(raw), banned) {
			t.Fatalf("%s: ingest stamps its own clock into the retained fact again, so no retry can be recognised", banned)
		}
	}
}