Merge feature/eng-04-connect-feld-mapping-konfiguration into integration/gesamtprojekt

connect/docs/ENG-04-PRUEFPROTOKOLL.md

@@ -0,0 +1,51 @@
+# ENG-04 – Feld-Mapping-Konfiguration (NEXARCH ↔ Drittsystem): Prüfprotokoll
+
+## Umsetzung
+
+Neues Paket `connect/internal/fieldmapping` (`Store`, `ValidateForExecution`),
+Migration `0003_field_mappings`. Jede Speicherung legt eine neue,
+unveränderliche Version an (`version` fortlaufend je Connector-Instanz),
+statt die vorherige zu überschreiben.
+
+Das Schema (welche NEXARCH-Felder es gibt und welche davon Pflicht sind)
+wird vom Connector-Plugin bereitgestellt (`[]FieldDef`), nicht in diesem
+Paket hartkodiert — ein neuer Connector mit eigenem Feldschema braucht
+damit keine Codeänderung an `fieldmapping`.
+
+## Prüfungen
+
+| # | Prüfung | Zielwert | Ergebnis |
+|---|---|---|---|
+| 1 | Ungültiges Mapping im Test abgewiesen | leer, unbekanntes Feld, leerer Name, doppeltes Quell- oder Zielfeld werden beim Speichern zurückgewiesen | `TestSave_RejectsInvalidMapping` (6 Unterfälle): bestanden |
+| 2 | Fehlendes Pflichtfeld-Mapping verhindert Sync-Start | `ValidateForExecution` meldet jedes fehlende Pflichtfeld namentlich, `nil` bei vollständiger Zuordnung | `TestValidateForExecution_ReportsMissingRequiredFields`: bestanden |
+| 3 | Geändertes Mapping wirkt erst auf nachfolgende Läufe | eine zu Laufbeginn erfasste Version bleibt unverändert lesbar, auch nachdem eine neue Version gespeichert wurde; neue Läufe sehen die neue Version | `TestGetVersion_ChangedMappingOnlyAffectsSubsequentRuns`: bestanden |
+
+Zusätzlich getestet: Konfiguration ohne Codeänderung möglich
+(`TestSave_ConfigurableWithoutCodeChange`), unbekannte Referenz liefert
+`ErrNotFound`.
+
+## Gesamtlauf auf 204
+
+- `go vet ./...`: sauber
+- `go test -count=1 ./internal/fieldmapping/...`: alle 5 Tests bestanden
+- `gofmt -l .`: leer (1 Korrektur: Struct-Feldausrichtung)
+
+## QA und Entscheidungen
+
+- **Versionierung statt Überschreiben**: Direkte Folge aus Prüfung 3 — ohne
+  unveränderliche Versionen ließe sich "wirkt erst auf nachfolgende Läufe"
+  nicht ohne einen separaten Snapshot-Mechanismus erfüllen. Eine Version zu
+  löschen ist bewusst nicht vorgesehen (kein Aufräum-Ticket im Scope).
+- **Schema kommt vom Aufrufer, nicht aus der Datenbank**: Das Ticket nennt
+  "Schema-Validierung" als technische Grundlage, aber keine eigene
+  Schema-Verwaltung als Akzeptanzkriterium. Ein Connector deklariert sein
+  Schema im Code (`[]FieldDef`); eine datenbankgestützte
+  Schema-Registrierung wäre vorzeitiger Umbau ohne Ticket-Deckung.
+- 1 Iteration: `gofmt`-Korrektur der Struct-Feldausrichtung.
+
+## Nicht Bestandteil dieses Tickets
+
+- Connect-Konfigurationsoberfläche (`API-01`) — dieses Paket ist reine
+  Backend-Logik, keine UI.
+- Transformationsregeln über reines Feld-zu-Feld-Mapping hinaus (das Ticket
+  nennt nur Feld-Mapping, keine Werttransformation).

connect/internal/fieldmapping/fieldmapping.go

@@ -0,0 +1,184 @@
+// Package fieldmapping manages the configurable field mapping between the
+// NEXARCH data model and a third-party connector's fields, per connector
+// instance (ENG-04).
+//
+// AC1: mapping is pure configuration (rows in field_mappings), no code
+// change needed to add or change a mapping.
+// AC2: Save validates against the connector's declared field schema and
+// rejects malformed mappings (unknown field, duplicate target, empty
+// names) before anything is written.
+// AC3: ValidateForExecution checks that every field the schema marks
+// required has a mapping entry, so a sync can be blocked before it starts
+// rather than failing mid-run on a missing field.
+package fieldmapping
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"sort"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+var (
+	ErrNotFound      = errors.New("fieldmapping: not found")
+	ErrInvalidConfig = errors.New("fieldmapping: invalid mapping")
+	ErrMissingFields = errors.New("fieldmapping: required field not mapped")
+)
+
+// Ref identifies the connector instance a mapping belongs to.
+type Ref struct {
+	TenantSlug          string
+	ConnectorName       string
+	ConnectorInstanceID string
+}
+
+// FieldDef declares one field of a connector's schema, provided by the
+// connector plugin (registry.Connector) rather than hard-coded here — that
+// is what lets a new connector be added without touching this package.
+type FieldDef struct {
+	NexarchField string
+	Required     bool
+}
+
+// Entry maps one NEXARCH field to one external field name.
+type Entry struct {
+	NexarchField  string `json:"nexarch_field"`
+	ExternalField string `json:"external_field"`
+}
+
+// Mapping is one saved, versioned configuration.
+type Mapping struct {
+	ID      string
+	Ref     Ref
+	Version int
+	Fields  []Entry
+}
+
+type Store struct {
+	pool *pgxpool.Pool
+}
+
+func NewStore(pool *pgxpool.Pool) *Store {
+	return &Store{pool: pool}
+}
+
+// validate checks fields for structural correctness and against the
+// connector's declared schema (AC2). It never touches the database.
+func validate(fields []Entry, schema []FieldDef) error {
+	if len(fields) == 0 {
+		return fmt.Errorf("%w: no fields", ErrInvalidConfig)
+	}
+	known := make(map[string]bool, len(schema))
+	for _, f := range schema {
+		known[f.NexarchField] = true
+	}
+	seenNexarch := make(map[string]bool, len(fields))
+	seenExternal := make(map[string]bool, len(fields))
+	for _, e := range fields {
+		if e.NexarchField == "" || e.ExternalField == "" {
+			return fmt.Errorf("%w: empty field name", ErrInvalidConfig)
+		}
+		if !known[e.NexarchField] {
+			return fmt.Errorf("%w: unknown NEXARCH field %q", ErrInvalidConfig, e.NexarchField)
+		}
+		if seenNexarch[e.NexarchField] {
+			return fmt.Errorf("%w: NEXARCH field %q mapped more than once", ErrInvalidConfig, e.NexarchField)
+		}
+		seenNexarch[e.NexarchField] = true
+		if seenExternal[e.ExternalField] {
+			return fmt.Errorf("%w: external field %q targeted more than once", ErrInvalidConfig, e.ExternalField)
+		}
+		seenExternal[e.ExternalField] = true
+	}
+	return nil
+}
+
+// Save validates fields against schema and, if valid, stores them as a new
+// version. It never overwrites a prior version (AC1, prepares Prüfung 3).
+func (s *Store) Save(ctx context.Context, ref Ref, fields []Entry, schema []FieldDef) (Mapping, error) {
+	if err := validate(fields, schema); err != nil {
+		return Mapping{}, err
+	}
+	payload, err := json.Marshal(fields)
+	if err != nil {
+		return Mapping{}, fmt.Errorf("fieldmapping: encode: %w", err)
+	}
+
+	var m Mapping
+	m.Ref = ref
+	m.Fields = fields
+	err = s.pool.QueryRow(ctx, `
+		INSERT INTO field_mappings (tenant_slug, connector_name, connector_instance_id, version, fields)
+		SELECT $1, $2, $3, COALESCE(MAX(version), 0) + 1, $4
+		FROM field_mappings
+		WHERE tenant_slug = $1 AND connector_name = $2 AND connector_instance_id = $3
+		RETURNING id, version
+	`, ref.TenantSlug, ref.ConnectorName, ref.ConnectorInstanceID, payload).Scan(&m.ID, &m.Version)
+	if err != nil {
+		return Mapping{}, fmt.Errorf("fieldmapping: save: %w", err)
+	}
+	return m, nil
+}
+
+// GetActive returns the highest (most recent) version for ref.
+func (s *Store) GetActive(ctx context.Context, ref Ref) (Mapping, error) {
+	return s.getBy(ctx, `
+		SELECT id, version, fields FROM field_mappings
+		WHERE tenant_slug = $1 AND connector_name = $2 AND connector_instance_id = $3
+		ORDER BY version DESC LIMIT 1
+	`, ref, ref.TenantSlug, ref.ConnectorName, ref.ConnectorInstanceID)
+}
+
+// GetVersion returns a specific, immutable version. A run that captured
+// this version at start time can keep reading it unchanged even after
+// Save creates a newer one (AC1 / Prüfung 3: a config change only affects
+// subsequent runs, not ones already using an older snapshot).
+func (s *Store) GetVersion(ctx context.Context, ref Ref, version int) (Mapping, error) {
+	return s.getBy(ctx, `
+		SELECT id, version, fields FROM field_mappings
+		WHERE tenant_slug = $1 AND connector_name = $2 AND connector_instance_id = $3 AND version = $4
+	`, ref, ref.TenantSlug, ref.ConnectorName, ref.ConnectorInstanceID, version)
+}
+
+func (s *Store) getBy(ctx context.Context, query string, ref Ref, args ...any) (Mapping, error) {
+	var m Mapping
+	m.Ref = ref
+	var payload []byte
+	err := s.pool.QueryRow(ctx, query, args...).Scan(&m.ID, &m.Version, &payload)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return Mapping{}, ErrNotFound
+	}
+	if err != nil {
+		return Mapping{}, fmt.Errorf("fieldmapping: get: %w", err)
+	}
+	if err := json.Unmarshal(payload, &m.Fields); err != nil {
+		return Mapping{}, fmt.Errorf("fieldmapping: decode: %w", err)
+	}
+	return m, nil
+}
+
+// ValidateForExecution checks m against schema's required fields (AC3):
+// every field schema marks required must have a mapping entry. It returns
+// ErrMissingFields naming every gap, so a sync can be blocked before it
+// starts.
+func ValidateForExecution(m Mapping, schema []FieldDef) error {
+	mapped := make(map[string]bool, len(m.Fields))
+	for _, e := range m.Fields {
+		mapped[e.NexarchField] = true
+	}
+	var missing []string
+	for _, f := range schema {
+		if f.Required && !mapped[f.NexarchField] {
+			missing = append(missing, f.NexarchField)
+		}
+	}
+	if len(missing) == 0 {
+		return nil
+	}
+	sort.Strings(missing)
+	return fmt.Errorf("%w: %v", ErrMissingFields, missing)
+}

connect/internal/fieldmapping/fieldmapping_test.go

@@ -0,0 +1,211 @@
+package fieldmapping
+
+import (
+	"context"
+	"errors"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"gitea.perlbach24.de/scripte/nexarch/connect/internal/migrate"
+	"gitea.perlbach24.de/scripte/nexarch/connect/internal/testdb"
+)
+
+func newStore(t *testing.T) *Store {
+	t.Helper()
+	dsn := os.Getenv("TEST_TENANT_DSN")
+	if dsn == "" {
+		t.Skip("TEST_TENANT_DSN nicht gesetzt, Integrationstest uebersprungen")
+	}
+	pool := testdb.Pool(t, dsn)
+
+	root, err := filepath.Abs("../../migrations")
+	if err != nil {
+		t.Fatalf("migrationsverzeichnis: %v", err)
+	}
+	migrations, err := migrate.Load(root)
+	if err != nil {
+		t.Fatalf("migrationen laden: %v", err)
+	}
+	if _, err := migrate.Up(context.Background(), pool, migrations); err != nil {
+		t.Fatalf("migrationen anwenden: %v", err)
+	}
+	return NewStore(pool)
+}
+
+func testRef(t *testing.T) Ref {
+	t.Helper()
+	return Ref{TenantSlug: "acme", ConnectorName: "testconnector", ConnectorInstanceID: t.Name()}
+}
+
+func testSchema() []FieldDef {
+	return []FieldDef{
+		{NexarchField: "title", Required: true},
+		{NexarchField: "created_by", Required: true},
+		{NexarchField: "tags", Required: false},
+	}
+}
+
+// AC1: Feld-Mapping ist ohne Codeänderung konfigurierbar.
+func TestSave_ConfigurableWithoutCodeChange(t *testing.T) {
+	s := newStore(t)
+	ctx := context.Background()
+
+	m, err := s.Save(ctx, testRef(t), []Entry{
+		{NexarchField: "title", ExternalField: "Subject"},
+		{NexarchField: "created_by", ExternalField: "Author"},
+	}, testSchema())
+	if err != nil {
+		t.Fatalf("save: %v", err)
+	}
+	if m.Version != 1 {
+		t.Fatalf("version = %d, want 1", m.Version)
+	}
+
+	got, err := s.GetActive(ctx, testRef(t))
+	if err != nil {
+		t.Fatalf("get active: %v", err)
+	}
+	if len(got.Fields) != 2 {
+		t.Fatalf("fields = %d, want 2", len(got.Fields))
+	}
+}
+
+// AC2 / Prüfung 1: ungültiges Mapping wird beim Speichern abgewiesen.
+func TestSave_RejectsInvalidMapping(t *testing.T) {
+	s := newStore(t)
+	ctx := context.Background()
+
+	cases := []struct {
+		name   string
+		fields []Entry
+	}{
+		{"leer", nil},
+		{"unbekanntes feld", []Entry{{NexarchField: "not_in_schema", ExternalField: "X"}}},
+		{"leerer external-name", []Entry{{NexarchField: "title", ExternalField: ""}}},
+		{"leerer nexarch-name", []Entry{{NexarchField: "", ExternalField: "X"}}},
+		{"doppeltes nexarch-feld", []Entry{
+			{NexarchField: "title", ExternalField: "Subject"},
+			{NexarchField: "title", ExternalField: "OtherSubject"},
+		}},
+		{"doppeltes ziel", []Entry{
+			{NexarchField: "title", ExternalField: "Subject"},
+			{NexarchField: "created_by", ExternalField: "Subject"},
+		}},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			ref := Ref{TenantSlug: "acme", ConnectorName: "testconnector", ConnectorInstanceID: t.Name()}
+			if _, err := s.Save(ctx, ref, c.fields, testSchema()); !errors.Is(err, ErrInvalidConfig) {
+				t.Fatalf("err = %v, want ErrInvalidConfig", err)
+			}
+		})
+	}
+}
+
+// AC3 / Prüfung 2: Pflichtfeld ohne Zuordnung wird vor Ausführung gemeldet.
+func TestValidateForExecution_ReportsMissingRequiredFields(t *testing.T) {
+	s := newStore(t)
+	ctx := context.Background()
+
+	// Nur "title" gemappt, "created_by" (required) fehlt.
+	m, err := s.Save(ctx, testRef(t), []Entry{
+		{NexarchField: "title", ExternalField: "Subject"},
+	}, testSchema())
+	if err != nil {
+		t.Fatalf("save: %v", err)
+	}
+
+	err = ValidateForExecution(m, testSchema())
+	if !errors.Is(err, ErrMissingFields) {
+		t.Fatalf("err = %v, want ErrMissingFields", err)
+	}
+	if !contains(err.Error(), "created_by") {
+		t.Fatalf("error %q does not name the missing field", err.Error())
+	}
+
+	// Vollständig gemappt: kein Fehler mehr.
+	m2, err := s.Save(ctx, testRef(t), []Entry{
+		{NexarchField: "title", ExternalField: "Subject"},
+		{NexarchField: "created_by", ExternalField: "Author"},
+	}, testSchema())
+	if err != nil {
+		t.Fatalf("save (complete): %v", err)
+	}
+	if err := ValidateForExecution(m2, testSchema()); err != nil {
+		t.Fatalf("validate for execution (complete mapping): %v", err)
+	}
+}
+
+// Prüfung 3: geändertes Mapping wirkt erst auf nachfolgende Läufe. Eine
+// bereits erfasste Mapping-Version (wie sie ein laufender Sync beim Start
+// festhalten würde) bleibt unverändert lesbar, auch nachdem eine neue
+// Version gespeichert wurde.
+func TestGetVersion_ChangedMappingOnlyAffectsSubsequentRuns(t *testing.T) {
+	s := newStore(t)
+	ctx := context.Background()
+	ref := testRef(t)
+
+	v1, err := s.Save(ctx, ref, []Entry{
+		{NexarchField: "title", ExternalField: "Subject"},
+		{NexarchField: "created_by", ExternalField: "Author"},
+	}, testSchema())
+	if err != nil {
+		t.Fatalf("save v1: %v", err)
+	}
+
+	// Ein "laufender Sync" erfasst v1 zu Beginn.
+	inFlight, err := s.GetVersion(ctx, ref, v1.Version)
+	if err != nil {
+		t.Fatalf("get version v1: %v", err)
+	}
+
+	// Konfiguration ändert sich während der in-flight-Sync noch läuft.
+	v2, err := s.Save(ctx, ref, []Entry{
+		{NexarchField: "title", ExternalField: "Headline"},
+		{NexarchField: "created_by", ExternalField: "Author"},
+	}, testSchema())
+	if err != nil {
+		t.Fatalf("save v2: %v", err)
+	}
+	if v2.Version != v1.Version+1 {
+		t.Fatalf("v2.Version = %d, want %d", v2.Version, v1.Version+1)
+	}
+
+	// Die bereits erfasste Snapshot-Version bleibt unverändert.
+	if inFlight.Fields[0].ExternalField != "Subject" {
+		t.Fatalf("in-flight snapshot wurde veraendert: %+v", inFlight.Fields)
+	}
+	reread, err := s.GetVersion(ctx, ref, v1.Version)
+	if err != nil {
+		t.Fatalf("get version v1 again: %v", err)
+	}
+	if reread.Fields[0].ExternalField != "Subject" {
+		t.Fatalf("gespeicherte v1 wurde veraendert: %+v", reread.Fields)
+	}
+
+	// Neue Läufe sehen die neue Version.
+	active, err := s.GetActive(ctx, ref)
+	if err != nil {
+		t.Fatalf("get active: %v", err)
+	}
+	if active.Version != v2.Version || active.Fields[0].ExternalField != "Headline" {
+		t.Fatalf("active mapping = %+v, want v2 mit Headline", active)
+	}
+}
+
+func TestGetActive_UnknownRef(t *testing.T) {
+	s := newStore(t)
+	if _, err := s.GetActive(context.Background(), testRef(t)); !errors.Is(err, ErrNotFound) {
+		t.Fatalf("err = %v, want ErrNotFound", err)
+	}
+}
+
+func contains(s, sub string) bool {
+	for i := 0; i+len(sub) <= len(s); i++ {
+		if s[i:i+len(sub)] == sub {
+			return true
+		}
+	}
+	return false
+}

connect/migrations/0003_field_mappings.down.sql

@@ -0,0 +1 @@
+DROP TABLE IF EXISTS field_mappings;

connect/migrations/0003_field_mappings.up.sql

@@ -0,0 +1,17 @@
+CREATE TABLE field_mappings (
+    id                     UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    tenant_slug            TEXT NOT NULL,
+    connector_name         TEXT NOT NULL,
+    connector_instance_id  TEXT NOT NULL,
+    version                INT NOT NULL,
+    fields                 JSONB NOT NULL,
+    created_at             TIMESTAMPTZ NOT NULL DEFAULT now(),
+    UNIQUE (tenant_slug, connector_name, connector_instance_id, version)
+);
+
+-- Jede Konfigurationsänderung legt eine neue Version an, statt die
+-- vorherige zu überschreiben: ein bereits gestarteter Lauf, der seine
+-- Mapping-Version erfasst hat, bleibt bei dieser Version, auch wenn danach
+-- eine neue Version gespeichert wird (Prüfung 3).
+CREATE INDEX idx_field_mappings_active
+    ON field_mappings (tenant_slug, connector_name, connector_instance_id, version DESC);