fluid-core/internal/observation/ingest_test.go

package observation

import (
	"context"
	"errors"
	"path/filepath"
	"strings"
	"testing"
	"time"

	_ "modernc.org/sqlite"

	"github.com/tegwick/fluid-core/internal/contract"
	"github.com/tegwick/fluid-core/internal/evidence"
)

func newIngest(t *testing.T) (*Ingest, *evidence.SQLStore) {
	t.Helper()
	store, err := evidence.OpenSQLite(context.Background(), filepath.Join(t.TempDir(), "e.db"))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = store.Close() })

	in, err := NewIngest(store, "hall-publishing", testPolicy())
	if err != nil {
		t.Fatal(err)
	}
	return in, store
}

func TestIngestRejectsUnsaltedPolicy(t *testing.T) {
	store, err := evidence.OpenSQLite(context.Background(), filepath.Join(t.TempDir(), "e.db"))
	if err != nil {
		t.Fatal(err)
	}
	defer store.Close()

	if _, err := NewIngest(store, "x", RedactionPolicy{}); !errors.Is(err, ErrNoSalt) {
		t.Errorf("an unsalted ingest was constructed: %v", err)
	}
}

// TestRedactionHappensOnTheWayIn: the store is append-only, so anything written
// unredacted stays that way forever.
func TestRedactionHappensOnTheWayIn(t *testing.T) {
	ctx := context.Background()
	in, store := newIngest(t)

	err := in.Write(ctx, contract.FluidTelemetry{
		ConsumerRef: "bernd@example.com",
		Request:     &contract.FluidTelemetryRequest{Route: "/v1/entries?token=hunter2", Method: "GET"},
	})
	if err != nil {
		t.Fatal(err)
	}

	rows, err := store.Telemetry(ctx, evidence.TelemetryFilter{})
	if err != nil {
		t.Fatal(err)
	}
	if len(rows) != 1 {
		t.Fatalf("stored %d rows, want 1", len(rows))
	}

	stored := rows[0]
	if strings.Contains(stored.ConsumerRef, "@") {
		t.Errorf("raw consumer identity reached the store: %q", stored.ConsumerRef)
	}
	if strings.Contains(stored.Request.Route, "hunter2") {
		t.Errorf("credential reached the store: %q", stored.Request.Route)
	}
	if stored.Redaction == nil || !stored.Redaction.Applied {
		t.Error("redaction was applied but not recorded on the stored event")
	}
}

// TestKindIsInferredFromShape: an error event filed as a request would
// understate the interface's failure rate.
func TestKindIsInferredFromShape(t *testing.T) {
	in, _ := newIngest(t)

	got, err := in.Normalize(contract.FluidTelemetry{
		Error: &contract.FluidTelemetryError{Class: contract.FluidTelemetryErrorClassTimeout},
	})
	if err != nil {
		t.Fatal(err)
	}
	if got.Kind != contract.FluidTelemetryKindError {
		t.Errorf("kind = %s, want error", got.Kind)
	}

	got, err = in.Normalize(contract.FluidTelemetry{
		Adoption: &contract.FluidTelemetryAdoption{Event: contract.FluidTelemetryAdoptionEventFirstUse},
	})
	if err != nil {
		t.Fatal(err)
	}
	if got.Kind != contract.FluidTelemetryKindAdoption {
		t.Errorf("kind = %s, want adoption", got.Kind)
	}
}

func TestNormalizeFillsDefaultsAndGuardsInterface(t *testing.T) {
	in, _ := newIngest(t)

	got, err := in.Normalize(contract.FluidTelemetry{})
	if err != nil {
		t.Fatal(err)
	}
	if got.ID == "" || got.OccurredAt.IsZero() || got.SchemaVersion == "" {
		t.Errorf("defaults not filled: %+v", got)
	}
	if got.InterfaceID != "hall-publishing" {
		t.Errorf("interface = %q", got.InterfaceID)
	}

	// Another interface's telemetry must not land in this evidence store.
	if _, err := in.Normalize(contract.FluidTelemetry{InterfaceID: "some-other-api"}); !errors.Is(err, ErrWrongInterface) {
		t.Errorf("foreign telemetry accepted: %v", err)
	}
}

// TestBatchKeepsGoodEventsWhenOneIsBad: dropping a hundred good events because
// one was malformed loses more than it protects.
func TestBatchKeepsGoodEventsWhenOneIsBad(t *testing.T) {
	ctx := context.Background()
	in, store := newIngest(t)

	accepted, rejected := in.WriteBatch(ctx, []contract.FluidTelemetry{
		{Request: &contract.FluidTelemetryRequest{Route: "/a"}},
		{InterfaceID: "wrong-interface"},
		{Request: &contract.FluidTelemetryRequest{Route: "/b"}},
	})

	if accepted != 2 {
		t.Errorf("accepted = %d, want 2", accepted)
	}
	if len(rejected) != 1 {
		t.Errorf("rejected = %d, want 1", len(rejected))
	}

	rows, _ := store.Telemetry(ctx, evidence.TelemetryFilter{})
	if len(rows) != 2 {
		t.Errorf("stored %d rows, want 2", len(rows))
	}
}

func TestErrorDetailIsBoundedWhenRawPayloadIsOff(t *testing.T) {
	in, _ := newIngest(t)

	long := strings.Repeat("x", 2000)
	got, err := in.Normalize(contract.FluidTelemetry{
		Error: &contract.FluidTelemetryError{
			Class:  contract.FluidTelemetryErrorClassBackendFailure,
			Detail: long,
		},
	})
	if err != nil {
		t.Fatal(err)
	}
	if len(got.Error.Detail) > 600 {
		t.Errorf("error detail was not bounded: %d chars", len(got.Error.Detail))
	}
}

func TestFeedbackIsStoredAsEvidence(t *testing.T) {
	ctx := context.Background()
	in, store := newIngest(t)

	cohort := contract.CohortID("coding-agents")
	got, err := in.RecordFeedback(ctx, contract.FluidFeedback{
		Cohort:            &cohort,
		Goal:              "publish a hall entry without splitting it by hand",
		MissingCapability: "long-form serialization",
		Outcome:           "capability unavailable, contact bernd@example.com",
	})
	if err != nil {
		t.Fatal(err)
	}

	if got.ID == "" || !strings.HasPrefix(string(got.ID), "F-") {
		t.Errorf("feedback id = %q, want an F- prefix", got.ID)
	}
	// Consumers write free text; it passes the same filter as everything else.
	if strings.Contains(got.Outcome, "bernd@example.com") {
		t.Errorf("an address survived in feedback: %q", got.Outcome)
	}

	if _, err := store.Record(ctx, contract.KindFeedback, string(got.ID)); err != nil {
		t.Errorf("feedback was not persisted: %v", err)
	}

	events, err := store.Events(ctx, evidence.EventFilter{EntityID: string(got.ID)})
	if err != nil {
		t.Fatal(err)
	}
	if len(events) != 1 {
		t.Fatalf("feedback left %d events, want 1", len(events))
	}
	// A consumer is untrusted; recording them as the actor keeps that visible.
	if events[0].Actor.Type != contract.ActorTypeConsumer {
		t.Errorf("actor type = %s, want consumer", events[0].Actor.Type)
	}
}

func TestFeedbackNeedsAGoal(t *testing.T) {
	in, _ := newIngest(t)
	// Without a goal there is nothing to interpret later.
	if _, err := in.RecordFeedback(context.Background(), contract.FluidFeedback{
		Outcome: "it did not work",
	}); err == nil {
		t.Error("feedback with no goal was accepted")
	}
}

// TestFeedbackDoesNotCreatePressure is the section 15 boundary: feedback is
// evidence and must not itself authorize a change.
func TestFeedbackDoesNotCreatePressure(t *testing.T) {
	ctx := context.Background()
	in, store := newIngest(t)

	if _, err := in.RecordFeedback(ctx, contract.FluidFeedback{
		Goal:              "I need a bulk publish endpoint",
		MissingCapability: "bulk publish",
	}); err != nil {
		t.Fatal(err)
	}

	records, err := store.Records(ctx, contract.KindPressure)
	if err != nil {
		t.Fatal(err)
	}
	if len(records) != 0 {
		t.Errorf("feedback created %d pressure records on its own", len(records))
	}
}

func TestIngestIsUsableAsARuntimeSink(t *testing.T) {
	ctx := context.Background()
	in, store := newIngest(t)

	// The gateway's emitter writes through this interface; redaction must apply
	// on that path too, not only to events arriving over the endpoint.
	var sink interface {
		Write(context.Context, contract.FluidTelemetry) error
	} = in

	if err := sink.Write(ctx, contract.FluidTelemetry{
		ConsumerRef: "raw-identity",
		OccurredAt:  time.Now(),
		Request:     &contract.FluidTelemetryRequest{Route: "/v1/x"},
	}); err != nil {
		t.Fatal(err)
	}

	rows, _ := store.Telemetry(ctx, evidence.TelemetryFilter{})
	if len(rows) != 1 {
		t.Fatalf("stored %d rows", len(rows))
	}
	if rows[0].ConsumerRef == "raw-identity" {
		t.Error("the emitter path bypassed redaction")
	}
}