nexarch/connect/internal/masterdatasync/masterdatasync_test.go

package masterdatasync

import (
	"context"
	"encoding/json"
	"io"
	"net/http"
	"net/http/httptest"
	"os"
	"path/filepath"
	"testing"
	"time"

	"github.com/jackc/pgx/v5/pgxpool"

	"gitea.perlbach24.de/scripte/nexarch/common/testdb"
	"gitea.perlbach24.de/scripte/nexarch/connect/internal/migrate"
	"gitea.perlbach24.de/scripte/nexarch/connect/internal/syncjob"
)

func newStore(t *testing.T) (*Store, *pgxpool.Pool) {
	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), pool
}

var testRef = Ref{TenantSlug: "acme", ConnectorName: "test-erp", ConnectorInstanceID: "inst-1"}

// listSource serves a fixed JSON array from a map, so a test can change
// what "the external system" returns between Reconcile calls.
type listSource struct {
	byEntity map[string][]byte
}

func (s *listSource) List(ctx context.Context, entity string) ([]byte, error) {
	return s.byEntity[entity], nil
}

func recordsJSON(t *testing.T, in []ExternalRecord) []byte {
	t.Helper()
	b, err := json.Marshal(in)
	if err != nil {
		t.Fatalf("marshal: %v", err)
	}
	return b
}

// jsonEqual compares two JSON documents by value, not by byte-for-byte
// formatting — Postgres re-serializes JSONB with its own whitespace on
// every read, so raw string comparison against a literal would be brittle.
func jsonEqual(t *testing.T, got json.RawMessage, wantLiteral string) bool {
	t.Helper()
	var a, b any
	if err := json.Unmarshal(got, &a); err != nil {
		t.Fatalf("jsonEqual: unmarshal got: %v", err)
	}
	if err := json.Unmarshal([]byte(wantLiteral), &b); err != nil {
		t.Fatalf("jsonEqual: unmarshal want: %v", err)
	}
	ga, _ := json.Marshal(a)
	gb, _ := json.Marshal(b)
	return string(ga) == string(gb)
}

// AC1: a record new on the external side appears in NEXARCH after sync.
func TestReconcile_NewExternalRecord_IsCreated(t *testing.T) {
	store, _ := newStore(t)
	src := &listSource{byEntity: map[string][]byte{
		"kunde": recordsJSON(t, []ExternalRecord{
			{ExternalID: "k-1", Data: json.RawMessage(`{"name":"Erstkunde"}`)},
		}),
	}}

	res, err := store.Reconcile(context.Background(), testRef, EntityKunde, src)
	if err != nil {
		t.Fatalf("reconcile: %v", err)
	}
	if res.Created != 1 || res.Updated != 0 || res.Conflicts != 0 {
		t.Fatalf("erwartet 1 created, 0 updated, 0 conflicts, bekam %+v", res)
	}

	rec, err := store.Get(context.Background(), testRef, EntityKunde, "k-1")
	if err != nil {
		t.Fatalf("get: %v", err)
	}
	if !jsonEqual(t, rec.NexarchData, `{"name":"Erstkunde"}`) {
		t.Fatalf("nexarch_data = %s, erwartet den externen Datensatz", rec.NexarchData)
	}
	if rec.Conflict {
		t.Fatalf("neuer Datensatz darf nicht als Konflikt markiert sein")
	}
}

// Ein zweiter Reconcile-Lauf ohne externe Aenderung ist ein No-Op (kein
// falscher "updated"-Zaehler).
func TestReconcile_NoChange_IsNoop(t *testing.T) {
	store, _ := newStore(t)
	src := &listSource{byEntity: map[string][]byte{
		"kunde": recordsJSON(t, []ExternalRecord{
			{ExternalID: "k-2", Data: json.RawMessage(`{"name":"Kunde Zwei"}`)},
		}),
	}}
	if _, err := store.Reconcile(context.Background(), testRef, EntityKunde, src); err != nil {
		t.Fatalf("erster reconcile: %v", err)
	}

	res, err := store.Reconcile(context.Background(), testRef, EntityKunde, src)
	if err != nil {
		t.Fatalf("zweiter reconcile: %v", err)
	}
	if res.Created != 0 || res.Updated != 0 || res.Conflicts != 0 {
		t.Fatalf("erwartet reines No-Op, bekam %+v", res)
	}
}

// Ein externe Aenderung ohne lokale Aenderung wird uebernommen.
func TestReconcile_ExternalChangeOnly_IsAdopted(t *testing.T) {
	store, _ := newStore(t)
	src := &listSource{byEntity: map[string][]byte{
		"kunde": recordsJSON(t, []ExternalRecord{
			{ExternalID: "k-3", Data: json.RawMessage(`{"name":"Alt"}`)},
		}),
	}}
	if _, err := store.Reconcile(context.Background(), testRef, EntityKunde, src); err != nil {
		t.Fatalf("erster reconcile: %v", err)
	}

	src.byEntity["kunde"] = recordsJSON(t, []ExternalRecord{
		{ExternalID: "k-3", Data: json.RawMessage(`{"name":"Neu"}`)},
	})
	res, err := store.Reconcile(context.Background(), testRef, EntityKunde, src)
	if err != nil {
		t.Fatalf("zweiter reconcile: %v", err)
	}
	if res.Updated != 1 || res.Conflicts != 0 {
		t.Fatalf("erwartet 1 updated, 0 conflicts, bekam %+v", res)
	}

	rec, err := store.Get(context.Background(), testRef, EntityKunde, "k-3")
	if err != nil {
		t.Fatalf("get: %v", err)
	}
	if !jsonEqual(t, rec.NexarchData, `{"name":"Neu"}`) {
		t.Fatalf("nexarch_data = %s, erwartet uebernommene externe Aenderung", rec.NexarchData)
	}
}

// AC2: gleichzeitige Aenderung auf beiden Seiten erzeugt einen Konflikt
// statt eines stillen Datenverlusts.
func TestReconcile_ConcurrentChangeBothSides_IsFlaggedNotOverwritten(t *testing.T) {
	store, _ := newStore(t)
	src := &listSource{byEntity: map[string][]byte{
		"kunde": recordsJSON(t, []ExternalRecord{
			{ExternalID: "k-4", Data: json.RawMessage(`{"name":"Ausgangsstand"}`)},
		}),
	}}
	if _, err := store.Reconcile(context.Background(), testRef, EntityKunde, src); err != nil {
		t.Fatalf("erster reconcile: %v", err)
	}

	// NEXARCH-seitige Aenderung, unabhaengig vom naechsten Sync.
	if err := store.SetLocalData(context.Background(), testRef, EntityKunde, "k-4", json.RawMessage(`{"name":"Lokal geaendert"}`)); err != nil {
		t.Fatalf("set local data: %v", err)
	}

	// Gleichzeitig hat sich auch die externe Seite geaendert.
	src.byEntity["kunde"] = recordsJSON(t, []ExternalRecord{
		{ExternalID: "k-4", Data: json.RawMessage(`{"name":"Extern geaendert"}`)},
	})

	res, err := store.Reconcile(context.Background(), testRef, EntityKunde, src)
	if err != nil {
		t.Fatalf("reconcile mit konflikt: %v", err)
	}
	if res.Conflicts != 1 || res.Updated != 0 {
		t.Fatalf("erwartet 1 conflict, 0 updated, bekam %+v", res)
	}

	rec, err := store.Get(context.Background(), testRef, EntityKunde, "k-4")
	if err != nil {
		t.Fatalf("get: %v", err)
	}
	if !rec.Conflict {
		t.Fatalf("record haette als conflict markiert sein muessen")
	}
	if !jsonEqual(t, rec.NexarchData, `{"name":"Lokal geaendert"}`) {
		t.Fatalf("nexarch_data wurde stillschweigend ueberschrieben: %s", rec.NexarchData)
	}
	if !jsonEqual(t, rec.ExternalData, `{"name":"Extern geaendert"}`) {
		t.Fatalf("external_data haette den neuen externen Stand tracken sollen: %s", rec.ExternalData)
	}
}

func TestResolveConflict_ClearsFlagAndAdoptsResolvedData(t *testing.T) {
	store, _ := newStore(t)
	src := &listSource{byEntity: map[string][]byte{
		"kunde": recordsJSON(t, []ExternalRecord{{ExternalID: "k-5", Data: json.RawMessage(`{"name":"A"}`)}}),
	}}
	if _, err := store.Reconcile(context.Background(), testRef, EntityKunde, src); err != nil {
		t.Fatalf("erster reconcile: %v", err)
	}
	if err := store.SetLocalData(context.Background(), testRef, EntityKunde, "k-5", json.RawMessage(`{"name":"lokal"}`)); err != nil {
		t.Fatalf("set local: %v", err)
	}
	src.byEntity["kunde"] = recordsJSON(t, []ExternalRecord{{ExternalID: "k-5", Data: json.RawMessage(`{"name":"extern"}`)}})
	if _, err := store.Reconcile(context.Background(), testRef, EntityKunde, src); err != nil {
		t.Fatalf("konflikt-reconcile: %v", err)
	}

	if err := store.ResolveConflict(context.Background(), testRef, EntityKunde, "k-5", json.RawMessage(`{"name":"aufgeloest"}`)); err != nil {
		t.Fatalf("resolve conflict: %v", err)
	}
	rec, err := store.Get(context.Background(), testRef, EntityKunde, "k-5")
	if err != nil {
		t.Fatalf("get: %v", err)
	}
	if rec.Conflict {
		t.Fatalf("conflict-flag haette geloescht sein sollen")
	}
	if !jsonEqual(t, rec.NexarchData, `{"name":"aufgeloest"}`) {
		t.Fatalf("nexarch_data = %s, erwartet aufgeloesten stand", rec.NexarchData)
	}

	// Ein weiterer Reconcile-Lauf mit unveraendertem externen Stand darf
	// den aufgeloesten Konflikt nicht wieder aufreissen.
	res, err := store.Reconcile(context.Background(), testRef, EntityKunde, src)
	if err != nil {
		t.Fatalf("reconcile nach resolve: %v", err)
	}
	if res.Conflicts != 0 {
		t.Fatalf("erwartet keinen erneuten conflict nach resolve, bekam %+v", res)
	}
}

// AC3: der Abgleich laeuft ueber die Sync-Job-Verwaltung nach Zeitplan.
func TestRunner_RunDueJobs_ExecutesReconcileOnSchedule(t *testing.T) {
	store, pool := newStore(t)
	jobs := syncjob.NewStore(pool)

	jobID, err := jobs.Create(context.Background(), syncjob.Ref{
		TenantSlug: testRef.TenantSlug, ConnectorName: testRef.ConnectorName, ConnectorInstanceID: testRef.ConnectorInstanceID,
	}, 60)
	if err != nil {
		t.Fatalf("create job: %v", err)
	}

	src := &listSource{byEntity: map[string][]byte{
		"kunde": recordsJSON(t, []ExternalRecord{{ExternalID: "k-6", Data: json.RawMessage(`{"name":"Zeitplan"}`)}}),
	}}
	runner := &Runner{Jobs: jobs, Store: store, Source: src, Entity: EntityKunde}

	// Der Job ist frisch angelegt und daher sofort faellig (next_run_at
	// wird bei Create auf "jetzt" gesetzt); ein Lauf in der nahen Zukunft
	// muss ihn also claimen und ausfuehren.
	n, err := runner.RunDueJobs(context.Background(), testRef, jobID, time.Now().Add(time.Second))
	if err != nil {
		t.Fatalf("run due jobs: %v", err)
	}
	if n != 1 {
		t.Fatalf("erwartet 1 ausgefuehrten job, bekam %d", n)
	}

	rec, err := store.Get(context.Background(), testRef, EntityKunde, "k-6")
	if err != nil {
		t.Fatalf("get nach scheduled run: %v", err)
	}
	if !jsonEqual(t, rec.NexarchData, `{"name":"Zeitplan"}`) {
		t.Fatalf("scheduled reconcile hat den datensatz nicht angelegt: %s", rec.NexarchData)
	}

	runs, err := jobs.ListRuns(context.Background(), jobID)
	if err != nil {
		t.Fatalf("list runs: %v", err)
	}
	if len(runs) != 1 || runs[0].Status != syncjob.RunStatusSucceeded {
		t.Fatalf("erwartet genau 1 erfolgreichen run, bekam %+v", runs)
	}
}

// AC1/AC2 end-to-end gegen einen echten HTTP-Testserver, nicht nur einen
// In-Memory-Source-Stub — belegt, dass Reconcile mit einem echten
// erpconnector.Client-artigen HTTP-Aufruf funktioniert.
func TestReconcile_AgainstRealHTTPSource(t *testing.T) {
	store, _ := newStore(t)
	body := `[{"ExternalID":"k-7","Data":{"name":"Http-Kunde"}}]`
	mux := http.NewServeMux()
	mux.HandleFunc("/kunde", func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "application/json")
		w.Write([]byte(body))
	})
	srv := httptest.NewServer(mux)
	t.Cleanup(srv.Close)

	src := &httpListSource{baseURL: srv.URL}
	res, err := store.Reconcile(context.Background(), testRef, EntityKunde, src)
	if err != nil {
		t.Fatalf("reconcile: %v", err)
	}
	if res.Created != 1 {
		t.Fatalf("erwartet 1 created ueber echten http-aufruf, bekam %+v", res)
	}
}

// httpListSource ruft testweise direkt einen echten HTTP-Endpunkt auf, im
// selben Format wie erpconnector.Client.List (GET /<entity>), ohne die
// Secret-Config von erpconnector.Config zu benoetigen.
type httpListSource struct{ baseURL string }

func (s *httpListSource) List(ctx context.Context, entity string) ([]byte, error) {
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, s.baseURL+"/"+entity, nil)
	if err != nil {
		return nil, err
	}
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()
	return io.ReadAll(resp.Body)
}