fluid-core/internal/observation/ingest.go

package observation

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"errors"
	"fmt"
	"time"

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

// Ingest normalizes, redacts and stores telemetry.
//
// Redaction happens here rather than at the query side because the evidence
// store is append-only: anything written unredacted stays unredacted forever.
// The filter belongs on the way in, where there is still a decision to make.
type Ingest struct {
	store  evidence.Store
	policy RedactionPolicy
	iface  contract.InterfaceID
	now    func() time.Time
}

// NewIngest returns an ingest pipeline.
func NewIngest(store evidence.Store, iface contract.InterfaceID, policy RedactionPolicy) (*Ingest, error) {
	if err := policy.Validate(); err != nil {
		return nil, err
	}
	return &Ingest{store: store, policy: policy, iface: iface, now: time.Now}, nil
}

// ErrWrongInterface reports telemetry submitted for another interface.
var ErrWrongInterface = errors.New("telemetry belongs to a different interface")

// Normalize fills in defaults and applies redaction.
//
// It is separated from Write so the same normalization runs whether an event
// arrives from the in-process emitter or over the ingest endpoint. Two paths
// with two normalizations would eventually disagree, and the disagreement would
// surface as a pressure finding that is really a bug in the pipeline.
func (i *Ingest) Normalize(ev contract.FluidTelemetry) (contract.FluidTelemetry, error) {
	if ev.InterfaceID == "" {
		ev.InterfaceID = i.iface
	}
	if ev.InterfaceID != i.iface {
		return ev, fmt.Errorf("%w: event is for %q, ingest serves %q",
			ErrWrongInterface, ev.InterfaceID, i.iface)
	}

	if ev.SchemaVersion == "" {
		ev.SchemaVersion = "0.1"
	}
	if ev.ID == "" {
		ev.ID = newID("tl-")
	}
	if ev.OccurredAt.IsZero() {
		ev.OccurredAt = i.now().UTC()
	}
	ev.OccurredAt = ev.OccurredAt.UTC()

	if ev.Kind == "" {
		// Infer from shape rather than defaulting to "request": an error event
		// filed as a request would understate the interface's failure rate.
		switch {
		case ev.Error != nil:
			ev.Kind = contract.FluidTelemetryKindError
		case ev.Adoption != nil:
			ev.Kind = contract.FluidTelemetryKindAdoption
		case ev.Sequence != nil:
			ev.Kind = contract.FluidTelemetryKindSequence
		default:
			ev.Kind = contract.FluidTelemetryKindRequest
		}
	}
	if !ev.Kind.Valid() {
		return ev, fmt.Errorf("unknown telemetry kind %q", ev.Kind)
	}

	// Raw payload capture is not the default (Blueprint 6.2). Where a policy
	// forbids it, error detail is the only free-text field that survives, and
	// it is scrubbed below.
	if !i.policy.AllowRawPayload && ev.Error != nil && ev.Error.Detail != "" {
		if len(ev.Error.Detail) > 512 {
			ev.Error.Detail = ev.Error.Detail[:512] + "…"
		}
	}

	i.policy.Apply(&ev)
	return ev, nil
}

// Write normalizes and stores one event.
func (i *Ingest) Write(ctx context.Context, ev contract.FluidTelemetry) error {
	normalized, err := i.Normalize(ev)
	if err != nil {
		return err
	}
	return i.store.WriteTelemetry(ctx, normalized)
}

// WriteBatch stores several events, reporting how many landed.
//
// A malformed event in a batch does not discard the rest. Telemetry is
// best-effort evidence, and dropping a hundred good events because one was
// wrong would lose more than it protects.
func (i *Ingest) WriteBatch(ctx context.Context, events []contract.FluidTelemetry) (accepted int, rejected []error) {
	for _, ev := range events {
		if err := i.Write(ctx, ev); err != nil {
			rejected = append(rejected, err)
			continue
		}
		accepted++
	}
	return accepted, rejected
}

// RecordFeedback stores explicit consumer feedback.
//
// FluidAPIStandards.md section 15: feedback is evidence and must not itself
// authorize interface changes. It is stored as a record and an event, and
// nothing here creates pressure or a hypothesis from it — that stays a
// deliberate step someone takes.
func (i *Ingest) RecordFeedback(ctx context.Context, f contract.FluidFeedback) (contract.FluidFeedback, error) {
	if f.Goal == "" {
		// Feedback with no stated goal cannot be interpreted later; the goal is
		// the part that says what the consumer was actually trying to do.
		return f, errors.New("feedback must state a goal")
	}

	if f.SchemaVersion == "" {
		f.SchemaVersion = "0.1"
	}
	if f.ID == "" {
		f.ID = contract.FeedbackID(newID("F-"))
	}
	if err := contract.RequireKind(string(f.ID), contract.KindFeedback); err != nil {
		return f, err
	}
	if f.InterfaceID == "" {
		f.InterfaceID = i.iface
	}
	if f.InterfaceID != i.iface {
		return f, fmt.Errorf("%w: feedback is for %q", ErrWrongInterface, f.InterfaceID)
	}
	if f.ReceivedAt.IsZero() {
		f.ReceivedAt = i.now().UTC()
	}

	// Consumers write free text; it passes the same filter as everything else.
	f.Goal, _ = i.policy.Scrub(f.Goal)
	f.Outcome, _ = i.policy.Scrub(f.Outcome)
	f.Attempt, _ = i.policy.Scrub(f.Attempt)
	f.MissingCapability, _ = i.policy.Scrub(f.MissingCapability)

	body, err := marshalFeedback(f)
	if err != nil {
		return f, err
	}
	if err := i.store.PutRecord(ctx, contract.KindFeedback, string(f.ID), body); err != nil {
		return f, err
	}

	if err := i.store.AppendEvent(ctx, contract.FluidEvent{
		SchemaVersion: "0.1",
		ID:            contract.EventID(fmt.Sprintf("EV-%s-%d", f.ID, i.now().UnixNano())),
		OccurredAt:    f.ReceivedAt,
		EntityType:    contract.FluidEventEntityTypeFeedback,
		EntityID:      string(f.ID),
		EventType:     "FEEDBACK_RECEIVED",
		// The consumer is the actor, and a consumer is untrusted (Blueprint 47).
		// Recording them as the actor keeps that visible in the audit trail.
		Actor:  contract.Actor{Type: contract.ActorTypeConsumer, ID: consumerLabel(f)},
		Reason: f.Goal,
	}); err != nil {
		return f, err
	}

	return f, nil
}

// newID generates an identifier with the given prefix.
func newID(prefix string) string {
	var b [10]byte
	if _, err := rand.Read(b[:]); err != nil {
		return prefix + "0"
	}
	return prefix + hex.EncodeToString(b[:])
}