tutti/packages/agent/activity-replication/conformance/conformance.go

// Package conformance provides backend-neutral activity replication fixtures.
// Projection builders and cloud sinks use separate harnesses because they sit
// on opposite sides of the replication boundary.
package conformance

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"reflect"

	activityreplication "github.com/tutti-os/tutti/packages/agent/activity-replication"
)

// ProjectionBuilder seeds canonical state and builds its wire projection. It
// is implemented by the Tutti SQLite adapter and the tsh local builder tests.
type ProjectionBuilder interface {
	Reset(context.Context) error
	Seed(context.Context, []CanonicalSeed) error
	Build(context.Context) (activityreplication.ChangeBatch, error)
}

// CanonicalSeed is source-side canonical state without mutation transport
// identity. Adapters persist these snapshots through their native store APIs.
type CanonicalSeed struct {
	WorkspaceID  string
	EntityType   activityreplication.EntityType
	Key          activityreplication.EntityKey
	Target       *activityreplication.Target
	TargetScope  *activityreplication.TargetScope
	Session      *activityreplication.Session
	SessionScope *activityreplication.SessionScope
	Turn         *activityreplication.Turn
	Interaction  *activityreplication.Interaction
	Message      *activityreplication.Message
}

type ProjectionFixture struct {
	Name      string
	Canonical []CanonicalSeed
	WantBatch activityreplication.ChangeBatch
}

// RunProjection verifies source-side canonical-to-wire projection. Mutation
// and transaction IDs are transport identities generated by each builder, so
// the comparison checks that they validate but ignores their concrete values.
func RunProjection(ctx context.Context, builder ProjectionBuilder, fixture ProjectionFixture) error {
	if err := builder.Reset(ctx); err != nil {
		return fmt.Errorf("reset projection builder: %w", err)
	}
	if err := builder.Seed(ctx, fixture.Canonical); err != nil {
		return fmt.Errorf("seed canonical state: %w", err)
	}
	got, err := builder.Build(ctx)
	if err != nil {
		return fmt.Errorf("build projection: %w", err)
	}
	if err := activityreplication.ValidateBatch(got); err != nil {
		return fmt.Errorf("built batch is invalid: %w", err)
	}
	if err := activityreplication.ValidateBatch(fixture.WantBatch); err != nil {
		return fmt.Errorf("fixture batch is invalid: %w", err)
	}
	retry, err := builder.Build(ctx)
	if err != nil {
		return fmt.Errorf("rebuild projection after lost response: %w", err)
	}
	if err := activityreplication.ValidateBatch(retry); err != nil {
		return fmt.Errorf("rebuilt batch is invalid: %w", err)
	}
	if len(retry.Mutations) != len(got.Mutations) {
		return fmt.Errorf("rebuilt batch has %d mutations, want %d", len(retry.Mutations), len(got.Mutations))
	}
	for index := range got.Mutations {
		if retry.Mutations[index].MutationID != got.Mutations[index].MutationID {
			return fmt.Errorf("rebuilt mutation %d changed mutation ID %q to %q", index, got.Mutations[index].MutationID, retry.Mutations[index].MutationID)
		}
	}
	if !reflect.DeepEqual(normalizeProjectionBatch(retry), normalizeProjectionBatch(got)) {
		return errors.New("rebuilt batch changed its canonical projection")
	}
	got = normalizeProjectionBatch(got)
	want := normalizeProjectionBatch(fixture.WantBatch)
	if !reflect.DeepEqual(got, want) {
		gotJSON, _ := json.Marshal(got)
		wantJSON, _ := json.Marshal(want)
		return fmt.Errorf("projection %s, want %s", gotJSON, wantJSON)
	}
	return nil
}

func normalizeProjectionBatch(batch activityreplication.ChangeBatch) activityreplication.ChangeBatch {
	batch.Mutations = append([]activityreplication.Mutation(nil), batch.Mutations...)
	for index := range batch.Mutations {
		batch.Mutations[index].MutationID = ""
		batch.Mutations[index].TransactionID = ""
	}
	return batch
}

type ApplyReport struct {
	Result           activityreplication.ApplyResult
	Acknowledgements []activityreplication.MutationAcknowledgement
}

// Sink is implemented by cloud projection sinks such as tsh-server MySQL.
type Sink interface {
	Reset(context.Context) error
	Apply(context.Context, activityreplication.ChangeBatch) (ApplyReport, error)
	Lookup(context.Context, activityreplication.EntityType, activityreplication.EntityKey) (SinkSnapshot, bool, error)
}

// SinkSnapshot is the persisted read model observed by the sink conformance
// harness. Scopes are explicit because acknowledgements and entity payloads
// alone cannot prove that routing and authorization attribution was retained.
type SinkSnapshot struct {
	Entity       json.RawMessage
	TargetScope  *activityreplication.TargetScope
	SessionScope *activityreplication.SessionScope
}

type SinkStep struct {
	Name                 string
	Batch                activityreplication.ChangeBatch
	WantResult           activityreplication.ApplyResult
	WantAcknowledgements []activityreplication.MutationAcknowledgement
	WantRejection        *RejectionExpectation
}

type RejectionExpectation struct {
	Kind          activityreplication.RejectionKind
	MutationID    string
	TransactionID string
}

type SnapshotExpectation struct {
	EntityType activityreplication.EntityType
	Key        activityreplication.EntityKey
	Present    bool
	Snapshot   SinkSnapshot
}

type SinkFixture struct {
	Name          string
	Steps         []SinkStep
	WantSnapshots []SnapshotExpectation
}

// RunSink verifies sink acknowledgements and final projection state. It checks
// each acknowledgement in full, including retry transaction identity and the
// original cursor returned for duplicates.
func RunSink(ctx context.Context, sink Sink, fixture SinkFixture) error {
	if err := sink.Reset(ctx); err != nil {
		return fmt.Errorf("reset sink: %w", err)
	}
	for _, step := range fixture.Steps {
		report, err := sink.Apply(ctx, step.Batch)
		if step.WantRejection != nil {
			if rejectionErr := matchRejection(err, *step.WantRejection); rejectionErr != nil {
				return fmt.Errorf("step %q: %w", step.Name, rejectionErr)
			}
			continue
		}
		if err != nil {
			return fmt.Errorf("step %q: apply: %w", step.Name, err)
		}
		summarized, err := activityreplication.SummarizeAcknowledgements(report.Acknowledgements)
		if err != nil {
			return fmt.Errorf("step %q: summarize acknowledgements: %w", step.Name, err)
		}
		if report.Result != summarized {
			return fmt.Errorf("step %q: result %#v does not summarize acknowledgements as %#v", step.Name, report.Result, summarized)
		}
		if report.Result != step.WantResult {
			return fmt.Errorf("step %q: result %#v, want %#v", step.Name, report.Result, step.WantResult)
		}
		if !reflect.DeepEqual(report.Acknowledgements, step.WantAcknowledgements) {
			return fmt.Errorf("step %q: acknowledgements %#v, want %#v", step.Name, report.Acknowledgements, step.WantAcknowledgements)
		}
	}
	for _, expectation := range fixture.WantSnapshots {
		got, found, err := sink.Lookup(ctx, expectation.EntityType, expectation.Key)
		if err != nil {
			return fmt.Errorf("lookup %s: %w", expectation.EntityType, err)
		}
		if found != expectation.Present {
			return fmt.Errorf("lookup %s: found %t, want %t", expectation.EntityType, found, expectation.Present)
		}
		if !expectation.Present {
			continue
		}
		if !jsonEqual(got.Entity, expectation.Snapshot.Entity) {
			return fmt.Errorf("lookup %s: entity snapshot %s, want %s", expectation.EntityType, got.Entity, expectation.Snapshot.Entity)
		}
		if !reflect.DeepEqual(got.TargetScope, expectation.Snapshot.TargetScope) {
			return fmt.Errorf("lookup %s: target scope %#v, want %#v", expectation.EntityType, got.TargetScope, expectation.Snapshot.TargetScope)
		}
		if !reflect.DeepEqual(got.SessionScope, expectation.Snapshot.SessionScope) {
			return fmt.Errorf("lookup %s: session scope %#v, want %#v", expectation.EntityType, got.SessionScope, expectation.Snapshot.SessionScope)
		}
	}
	return nil
}

func matchRejection(err error, want RejectionExpectation) error {
	if err == nil {
		return errors.New("expected permanent rejection, got nil")
	}
	var rejection *activityreplication.PermanentRejection
	if !errors.As(err, &rejection) {
		return fmt.Errorf("error %T is not a permanent rejection: %w", err, err)
	}
	if rejection.Kind != want.Kind || rejection.MutationID != want.MutationID || rejection.TransactionID != want.TransactionID {
		return fmt.Errorf("rejection %#v, want %#v", rejection, want)
	}
	return nil
}

func jsonEqual(left, right json.RawMessage) bool {
	var leftValue any
	var rightValue any
	return json.Unmarshal(left, &leftValue) == nil && json.Unmarshal(right, &rightValue) == nil && reflect.DeepEqual(leftValue, rightValue)
}