diff --git a/connect/docs/ERP-04-PRUEFPROTOKOLL.md b/connect/docs/ERP-04-PRUEFPROTOKOLL.md new file mode 100644 index 00000000..1a685d70 --- /dev/null +++ b/connect/docs/ERP-04-PRUEFPROTOKOLL.md @@ -0,0 +1,51 @@ +# ERP-04 – Prüfprotokoll + +Beleg-/Dokumentexport an ERP, aufbauend auf ERP-01 (`connect/internal/erpconnector`) und ENG-04 (`connect/internal/fieldmapping`). + +## Einschränkung: kein echtes ERP-System verfügbar + +Wie bei M365-01/DMS2-03 gibt es kein reales angebundenes ERP-System für Tests. Alle Prüfungen laufen gegen einen In-Prozess-Referenzserver (`newReferenceERP` in `connect/internal/erpexport/store_test.go`), der das dokumentierte Vertragsverhalten von `erpconnector.Client.Write` nachbildet (POST `/beleg`, JSON-Body, 401/403 = permanenter Fehler, sonstige Nicht-2xx = retrybar). Kein produktives ERP wurde angesprochen. + +## Umsetzung + +- `connect/internal/erpexport/erpexport.go`: `BuildPayload` rendert die Beleg-Metadaten über eine gespeicherte Feld-Mapping-Konfiguration (ENG-04) in die externen Feldnamen des ERP. +- `connect/internal/erpexport/store.go`: `Store` (Postgres, Tabelle `erp_exports`) exportiert einen Beleg über `erpconnector.Client.Write` (Entity `"beleg"`), mit Exponential-Backoff-Retry (gleiche Policy-Form wie `datevtransfer`) und Deduplizierung über `UNIQUE (tenant_slug, connector_instance_id, beleg_id)`. +- `connect/internal/erpconnector/erpconnector.go`: `Write` klassifiziert 401/403 zusätzlich als `ErrPermanent`, damit Retry-Logik permanente von vorübergehenden Fehlern unterscheiden kann (minimale Erweiterung der Vorbedingung, keine Verhaltensänderung für bestehende Aufrufer). +- Migration `connect/migrations/0010_erp_exports.{up,down}.sql`. + +## Prüfungen + +### 1. Testbeleg im ERP nach Export mit korrekten Metadaten auffindbar + +- **Ziel:** Ein Beleg mit zugeordneten Metadaten lässt sich exportieren; das Zielsystem erhält die per Feld-Mapping übersetzten Werte unverändert. +- **Test:** `TestStore_Start_SuccessWithCorrectMetadata` (`connect/internal/erpexport/store_test.go`). +- **Ergebnis:** PASS. Referenzserver empfängt genau `{"InvoiceNo":"RE-2026-001","Amount":"199.00"}`, Export-Status `success`, `ERPRef` gesetzt. + +### 2. Simulierter Exportfehler löst Retry aus + +- **Ziel:** Ein vorübergehender Fehler (503) führt zu `retry_pending` mit zukünftigem `retry_at` statt sofortigem `failed`; ein anschließender Retry gegen das inzwischen funktionierende System schließt erfolgreich ab. Ein permanenter Fehler (401) wird dagegen nicht retryed. +- **Tests:** `TestStore_Start_TransientFailureSchedulesRetryThenSucceeds`, `TestStore_Start_PermanentFailureDoesNotRetry`. +- **Ergebnis:** PASS für beide. Transient: Status `retry_pending` → nach `Retry()` `success`, 2 Versuche protokolliert. Permanent: sofort `failed`, kein `retry_at`. + +### 3. Doppelversand im Test ausgeschlossen + +- **Ziel:** Ein bereits erfolgreich exportierter Beleg wird bei erneutem Aufruf von `Start` nicht ein zweites Mal an das ERP übertragen. +- **Test:** `TestStore_Start_AlreadyExportedBelegNotSentTwice`. +- **Ergebnis:** PASS. Zweiter `Start`-Aufruf mit derselben Beleg-ID liefert denselben `Export` (gleiche ID, gleiche `ERPRef`) zurück; der Referenzserver zeichnet nur 1 Übertragung auf. + +### Ergänzend + +- `TestStore_ClaimDueRetries_OnlyReturnsDueExports`: fällige Retries werden korrekt gefunden, nicht fällige nicht. PASS. +- `TestBuildPayload_RendersMappedFields`, `TestBuildPayload_MissingFieldRejected`: Feld-Mapping-Rendering und Ablehnung bei fehlendem Beleg-Feld. PASS. + +## Testlauf + +``` +cd connect +TEST_TENANT_DSN=postgres://erp04_test:erp04_test_pw@localhost:5432/erp04_test?sslmode=disable \ + go test ./internal/erpexport/... ./internal/erpconnector/... -v +``` + +Alle 13 Tests PASS. `go build ./...` und `go vet ./...` (im Modul `connect/`) sind fehlerfrei. + +Testdatenbank `erp04_test` (Rolle `erp04_test`) wurde eigens für diese Kachel auf Host 204 angelegt, kein Zugriff auf `archivdms` oder Host 131. diff --git a/connect/internal/erpconnector/erpconnector.go b/connect/internal/erpconnector/erpconnector.go index e876d07c..2b421225 100644 --- a/connect/internal/erpconnector/erpconnector.go +++ b/connect/internal/erpconnector/erpconnector.go @@ -24,6 +24,10 @@ import ( // abort or a bare network error. var ErrConnectionFailed = errors.New("erpconnector: connection failed") +// ErrPermanent wraps a Write failure that must never be retried (401/403: +// retrying with the same credentials would never succeed). +var ErrPermanent = errors.New("erpconnector: permanent failure") + // Config configures one ERP connector instance. BaseURL and HealthPath are // read from environment variables by the caller, never hardcoded. type Config struct { @@ -126,6 +130,9 @@ func (c *Client) Write(ctx context.Context, entity string, data []byte) (string, return "", fmt.Errorf("%w: %w", ErrConnectionFailed, err) } defer resp.Body.Close() + if resp.StatusCode == http.StatusUnauthorized || resp.StatusCode == http.StatusForbidden { + return "", fmt.Errorf("%w: %w: status %d", ErrConnectionFailed, ErrPermanent, resp.StatusCode) + } if resp.StatusCode < 200 || resp.StatusCode >= 300 { return "", fmt.Errorf("%w: status %d", ErrConnectionFailed, resp.StatusCode) } diff --git a/connect/internal/erpexport/erpexport.go b/connect/internal/erpexport/erpexport.go new file mode 100644 index 00000000..d4a065fa --- /dev/null +++ b/connect/internal/erpexport/erpexport.go @@ -0,0 +1,39 @@ +// Package erpexport implements ERP-04: export of a Beleg with its mapped +// metadata to the connected ERP system, on top of ERP-01's erpconnector +// client and ENG-04's fieldmapping configuration. +// +// AC1: BuildPayload renders a Beleg's fields through a saved field mapping +// (ENG-04) into the ERP's external field names before Store.Start writes +// it via the entity "beleg". +// AC2: a failed export attempt is retried with backoff (same policy shape +// as datevtransfer); 401/403 from the ERP is treated as permanent and not +// retried. +// AC3: Store.Start is keyed on (tenant, connector instance, Beleg-ID) so a +// Beleg that already exported successfully is never sent a second time. +package erpexport + +import ( + "encoding/json" + "errors" + "fmt" + + "gitea.perlbach24.de/scripte/nexarch/connect/internal/fieldmapping" +) + +// ErrMissingBelegField signals that a mapped NEXARCH field has no value +// in the Beleg being exported. +var ErrMissingBelegField = errors.New("erpexport: beleg field not set") + +// BuildPayload renders belegFields through m into the ERP's external +// field names (AC1). +func BuildPayload(m fieldmapping.Mapping, belegFields map[string]string) ([]byte, error) { + out := make(map[string]string, len(m.Fields)) + for _, e := range m.Fields { + v, ok := belegFields[e.NexarchField] + if !ok { + return nil, fmt.Errorf("%w: %s", ErrMissingBelegField, e.NexarchField) + } + out[e.ExternalField] = v + } + return json.Marshal(out) +} diff --git a/connect/internal/erpexport/erpexport_test.go b/connect/internal/erpexport/erpexport_test.go new file mode 100644 index 00000000..793d64bb --- /dev/null +++ b/connect/internal/erpexport/erpexport_test.go @@ -0,0 +1,42 @@ +package erpexport + +import ( + "encoding/json" + "errors" + "testing" + + "gitea.perlbach24.de/scripte/nexarch/connect/internal/fieldmapping" +) + +// AC1: BuildPayload renders a Beleg's NEXARCH fields into the ERP's +// external field names per the saved mapping. +func TestBuildPayload_RendersMappedFields(t *testing.T) { + m := fieldmapping.Mapping{Fields: []fieldmapping.Entry{ + {NexarchField: "belegnummer", ExternalField: "InvoiceNo"}, + {NexarchField: "betrag", ExternalField: "Amount"}, + }} + payload, err := BuildPayload(m, map[string]string{ + "belegnummer": "RE-2026-001", + "betrag": "199.00", + }) + if err != nil { + t.Fatalf("BuildPayload: %v", err) + } + var got map[string]string + if err := json.Unmarshal(payload, &got); err != nil { + t.Fatalf("payload nicht lesbar: %v", err) + } + if got["InvoiceNo"] != "RE-2026-001" || got["Amount"] != "199.00" { + t.Fatalf("payload = %v, unerwartet", got) + } +} + +func TestBuildPayload_MissingFieldRejected(t *testing.T) { + m := fieldmapping.Mapping{Fields: []fieldmapping.Entry{ + {NexarchField: "belegnummer", ExternalField: "InvoiceNo"}, + }} + _, err := BuildPayload(m, map[string]string{}) + if !errors.Is(err, ErrMissingBelegField) { + t.Fatalf("erwartete ErrMissingBelegField, hatte %v", err) + } +} diff --git a/connect/internal/erpexport/store.go b/connect/internal/erpexport/store.go new file mode 100644 index 00000000..3e80fd15 --- /dev/null +++ b/connect/internal/erpexport/store.go @@ -0,0 +1,215 @@ +package erpexport + +import ( + "context" + "errors" + "fmt" + "math" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" + + "gitea.perlbach24.de/scripte/nexarch/connect/internal/erpconnector" +) + +// RetryPolicy controls exponential backoff between failed export +// attempts (AC2). Same shape as connect/internal/datevtransfer's +// RetryPolicy, kept separate since erpexport has its own status model. +type RetryPolicy struct { + MaxAttempts int + BaseDelay time.Duration + MaxDelay time.Duration +} + +func (p RetryPolicy) delay(attempt int) time.Duration { + d := p.BaseDelay * time.Duration(math.Pow(2, float64(attempt-1))) + if d > p.MaxDelay { + return p.MaxDelay + } + return d +} + +var DefaultRetryPolicy = RetryPolicy{MaxAttempts: 5, BaseDelay: 30 * time.Second, MaxDelay: 30 * time.Minute} + +type Status string + +const ( + StatusPending Status = "pending" + StatusSuccess Status = "success" + StatusFailed Status = "failed" + StatusRetryPending Status = "retry_pending" +) + +// Ref identifies the connector instance a Beleg is exported through. +type Ref struct { + TenantSlug string + ConnectorInstanceID string +} + +// Export is one persisted Beleg export attempt. +type Export struct { + ID string + Ref Ref + BelegID string + Status Status + Attempt int + LastError string + RetryAt *time.Time + ERPRef string + CreatedAt time.Time + UpdatedAt time.Time +} + +var ErrNotFound = errors.New("erpexport: export nicht gefunden") + +// Writer is the minimal capability Store needs from a connector: +// erpconnector.Client implements it via its Write method. +type Writer interface { + Write(ctx context.Context, entity string, data []byte) (string, error) +} + +type Store struct { + Pool *pgxpool.Pool + Writer Writer +} + +// Start exports belegID with payload for the first time. If belegID +// already has an export row for ref, it is not sent again: a prior +// success is returned as-is (AC3), while a prior failure/retry_pending +// row is retried instead of creating a duplicate. +func (s *Store) Start(ctx context.Context, ref Ref, belegID string, payload []byte) (Export, error) { + var e Export + err := s.Pool.QueryRow(ctx, ` + INSERT INTO erp_exports (tenant_slug, connector_instance_id, beleg_id, status, attempt) + VALUES ($1, $2, $3, $4, 0) + ON CONFLICT (tenant_slug, connector_instance_id, beleg_id) DO NOTHING + RETURNING id, tenant_slug, connector_instance_id, beleg_id, status, attempt, coalesce(last_error, ''), retry_at, coalesce(erp_ref, ''), created_at, updated_at + `, ref.TenantSlug, ref.ConnectorInstanceID, belegID, StatusPending). + Scan(&e.ID, &e.Ref.TenantSlug, &e.Ref.ConnectorInstanceID, &e.BelegID, &e.Status, &e.Attempt, &e.LastError, &e.RetryAt, &e.ERPRef, &e.CreatedAt, &e.UpdatedAt) + if errors.Is(err, pgx.ErrNoRows) { + existing, gerr := s.getByBeleg(ctx, ref, belegID) + if gerr != nil { + return Export{}, gerr + } + if existing.Status == StatusSuccess { + return existing, nil + } + return s.attempt(ctx, existing.ID, payload, DefaultRetryPolicy, time.Now()) + } + if err != nil { + return Export{}, fmt.Errorf("erpexport: export anlegen: %w", err) + } + return s.attempt(ctx, e.ID, payload, DefaultRetryPolicy, time.Now()) +} + +// Retry re-attempts an export that is due (status retry_pending). Callers +// pass the same payload since Postgres does not store the rendered +// bytes — a background worker would re-render via BuildPayload from the +// export's stored mapping version in a real deployment. +func (s *Store) Retry(ctx context.Context, exportID string, payload []byte) (Export, error) { + return s.attempt(ctx, exportID, payload, DefaultRetryPolicy, time.Now()) +} + +func (s *Store) attempt(ctx context.Context, exportID string, payload []byte, policy RetryPolicy, now time.Time) (Export, error) { + erpRef, writeErr := s.Writer.Write(ctx, "beleg", payload) + + var e Export + if writeErr == nil { + err := s.Pool.QueryRow(ctx, ` + UPDATE erp_exports + SET status = $2, attempt = attempt + 1, erp_ref = $3, last_error = NULL, retry_at = NULL, updated_at = now() + WHERE id = $1 + RETURNING id, tenant_slug, connector_instance_id, beleg_id, status, attempt, coalesce(last_error, ''), retry_at, coalesce(erp_ref, ''), created_at, updated_at + `, exportID, StatusSuccess, erpRef). + Scan(&e.ID, &e.Ref.TenantSlug, &e.Ref.ConnectorInstanceID, &e.BelegID, &e.Status, &e.Attempt, &e.LastError, &e.RetryAt, &e.ERPRef, &e.CreatedAt, &e.UpdatedAt) + if err != nil { + return Export{}, fmt.Errorf("erpexport: erfolg protokollieren: %w", err) + } + return e, nil + } + + current, err := s.Get(ctx, exportID) + if err != nil { + return Export{}, err + } + nextAttempt := current.Attempt + 1 + + permanent := errors.Is(writeErr, erpconnector.ErrPermanent) + status := StatusRetryPending + var retryAt *time.Time + if permanent || nextAttempt >= policy.MaxAttempts { + status = StatusFailed + } else { + at := now.Add(policy.delay(nextAttempt)) + retryAt = &at + } + + err = s.Pool.QueryRow(ctx, ` + UPDATE erp_exports + SET status = $2, attempt = $3, last_error = $4, retry_at = $5, updated_at = now() + WHERE id = $1 + RETURNING id, tenant_slug, connector_instance_id, beleg_id, status, attempt, coalesce(last_error, ''), retry_at, coalesce(erp_ref, ''), created_at, updated_at + `, exportID, status, nextAttempt, writeErr.Error(), retryAt). + Scan(&e.ID, &e.Ref.TenantSlug, &e.Ref.ConnectorInstanceID, &e.BelegID, &e.Status, &e.Attempt, &e.LastError, &e.RetryAt, &e.ERPRef, &e.CreatedAt, &e.UpdatedAt) + if err != nil { + return Export{}, fmt.Errorf("erpexport: fehlschlag protokollieren: %w", err) + } + return e, nil +} + +// Get returns one export's current status. +func (s *Store) Get(ctx context.Context, exportID string) (Export, error) { + var e Export + err := s.Pool.QueryRow(ctx, ` + SELECT id, tenant_slug, connector_instance_id, beleg_id, status, attempt, coalesce(last_error, ''), retry_at, coalesce(erp_ref, ''), created_at, updated_at + FROM erp_exports WHERE id = $1 + `, exportID). + Scan(&e.ID, &e.Ref.TenantSlug, &e.Ref.ConnectorInstanceID, &e.BelegID, &e.Status, &e.Attempt, &e.LastError, &e.RetryAt, &e.ERPRef, &e.CreatedAt, &e.UpdatedAt) + if errors.Is(err, pgx.ErrNoRows) { + return Export{}, ErrNotFound + } + if err != nil { + return Export{}, fmt.Errorf("erpexport: export lesen: %w", err) + } + return e, nil +} + +func (s *Store) getByBeleg(ctx context.Context, ref Ref, belegID string) (Export, error) { + var e Export + err := s.Pool.QueryRow(ctx, ` + SELECT id, tenant_slug, connector_instance_id, beleg_id, status, attempt, coalesce(last_error, ''), retry_at, coalesce(erp_ref, ''), created_at, updated_at + FROM erp_exports WHERE tenant_slug = $1 AND connector_instance_id = $2 AND beleg_id = $3 + `, ref.TenantSlug, ref.ConnectorInstanceID, belegID). + Scan(&e.ID, &e.Ref.TenantSlug, &e.Ref.ConnectorInstanceID, &e.BelegID, &e.Status, &e.Attempt, &e.LastError, &e.RetryAt, &e.ERPRef, &e.CreatedAt, &e.UpdatedAt) + if errors.Is(err, pgx.ErrNoRows) { + return Export{}, ErrNotFound + } + if err != nil { + return Export{}, fmt.Errorf("erpexport: export lesen: %w", err) + } + return e, nil +} + +// ClaimDueRetries returns every retry_pending export whose retry_at has +// passed — the building block for a future background retry worker (none +// exists yet, same documented-limitation pattern as datevtransfer). +func (s *Store) ClaimDueRetries(ctx context.Context, now time.Time) ([]Export, error) { + rows, err := s.Pool.Query(ctx, ` + SELECT id, tenant_slug, connector_instance_id, beleg_id, status, attempt, coalesce(last_error, ''), retry_at, coalesce(erp_ref, ''), created_at, updated_at + FROM erp_exports WHERE status = $1 AND retry_at <= $2 + `, StatusRetryPending, now) + if err != nil { + return nil, fmt.Errorf("erpexport: faellige retries lesen: %w", err) + } + defer rows.Close() + var out []Export + for rows.Next() { + var e Export + if err := rows.Scan(&e.ID, &e.Ref.TenantSlug, &e.Ref.ConnectorInstanceID, &e.BelegID, &e.Status, &e.Attempt, &e.LastError, &e.RetryAt, &e.ERPRef, &e.CreatedAt, &e.UpdatedAt); err != nil { + return nil, fmt.Errorf("erpexport: retry-zeile lesen: %w", err) + } + out = append(out, e) + } + return out, rows.Err() +} diff --git a/connect/internal/erpexport/store_test.go b/connect/internal/erpexport/store_test.go new file mode 100644 index 00000000..84a49892 --- /dev/null +++ b/connect/internal/erpexport/store_test.go @@ -0,0 +1,259 @@ +package erpexport + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "os" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + + "gitea.perlbach24.de/scripte/nexarch/connect/internal/credential" + "gitea.perlbach24.de/scripte/nexarch/connect/internal/erpconnector" + "gitea.perlbach24.de/scripte/nexarch/connect/internal/migrate" + "gitea.perlbach24.de/scripte/nexarch/connect/internal/testdb" +) + +func openTestPool(t *testing.T) *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) + + migrations, err := migrate.Load(findMigrationsDir(t)) + 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 pool +} + +func findMigrationsDir(t *testing.T) string { + t.Helper() + dir, _ := os.Getwd() + for i := 0; i < 6; i++ { + p := dir + "/migrations" + if fi, err := os.Stat(p); err == nil && fi.IsDir() { + return p + } + dir += "/.." + } + t.Fatal("connect/migrations nicht gefunden") + return "" +} + +// newReferenceERP is a minimal mock of a "test ERP system": there is no +// real ERP available for this ticket, so, like M365-01/DMS2-03, the +// AC/Pruefungen are verified against this in-process reference server +// instead — documented in ERP-04-PRUEFPROTOKOLL.md. It stores every +// accepted Beleg body in received so a test can assert the metadata +// arrived correctly (Pruefung 1), and fails the first failFirstN calls +// to POST /beleg with 503 to simulate a transient outage (Pruefung 2). +func newReferenceERP(t *testing.T, failFirstN int, permanentAuthFailure bool) (*httptest.Server, *[]map[string]string) { + t.Helper() + calls := 0 + var received []map[string]string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/health" { + w.WriteHeader(http.StatusOK) + return + } + if r.URL.Path != "/beleg" || r.Method != http.MethodPost { + http.NotFound(w, r) + return + } + calls++ + if permanentAuthFailure { + w.WriteHeader(http.StatusUnauthorized) + return + } + if calls <= failFirstN { + w.WriteHeader(http.StatusServiceUnavailable) + return + } + var body map[string]string + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + w.WriteHeader(http.StatusBadRequest) + return + } + received = append(received, body) + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]string{"id": fmt.Sprintf("erp-beleg-%d", calls)}) + })) + return srv, &received +} + +func testWriter(t *testing.T, endpoint string) Writer { + t.Helper() + return erpconnector.New(erpconnector.Config{ + Name: "test-erp", Version: "1.0.0", BaseURL: endpoint, HealthPath: "/health", + Secret: credential.Secret{APIKey: "test-key"}, + }) +} + +// AC1 / Pruefung 1: Beleg mit zugeordneten Metadaten wird exportiert und +// ist im (Referenz-)ERP mit korrekten Metadaten auffindbar. +func TestStore_Start_SuccessWithCorrectMetadata(t *testing.T) { + pool := openTestPool(t) + srv, received := newReferenceERP(t, 0, false) + defer srv.Close() + + s := &Store{Pool: pool, Writer: testWriter(t, srv.URL)} + ref := Ref{TenantSlug: "tenant-a", ConnectorInstanceID: "erp-a"} + payload := []byte(`{"InvoiceNo":"RE-2026-001","Amount":"199.00"}`) + + exp, err := s.Start(context.Background(), ref, "beleg-1", payload) + if err != nil { + t.Fatalf("Start: %v", err) + } + if exp.Status != StatusSuccess { + t.Fatalf("erwartete Status success, hatte %s", exp.Status) + } + if exp.ERPRef == "" { + t.Fatal("erwartete ERPRef") + } + if len(*received) != 1 || (*received)[0]["InvoiceNo"] != "RE-2026-001" || (*received)[0]["Amount"] != "199.00" { + t.Fatalf("erwartete korrekt uebertragene Metadaten im Referenz-ERP, hatte %v", *received) + } +} + +// AC2 / Pruefung 2: ein simulierter Exportfehler (503 einmalig) loest +// automatisch einen Retry aus statt sofort auf failed zu landen. +func TestStore_Start_TransientFailureSchedulesRetryThenSucceeds(t *testing.T) { + pool := openTestPool(t) + srv, _ := newReferenceERP(t, 1, false) + defer srv.Close() + + s := &Store{Pool: pool, Writer: testWriter(t, srv.URL)} + ref := Ref{TenantSlug: "tenant-b", ConnectorInstanceID: "erp-b"} + payload := []byte(`{"InvoiceNo":"RE-2026-002"}`) + + exp, err := s.Start(context.Background(), ref, "beleg-2", payload) + if err != nil { + t.Fatalf("Start: %v", err) + } + if exp.Status != StatusRetryPending { + t.Fatalf("erwartete Status retry_pending, hatte %s", exp.Status) + } + if exp.RetryAt == nil || !exp.RetryAt.After(time.Now()) { + t.Fatal("erwartete zukuenftiges retry_at") + } + if exp.LastError == "" { + t.Fatal("erwartete protokollierten Fehler") + } + + exp, err = s.Retry(context.Background(), exp.ID, payload) + if err != nil { + t.Fatalf("Retry: %v", err) + } + if exp.Status != StatusSuccess { + t.Fatalf("erwartete Status success nach Retry, hatte %s", exp.Status) + } + if exp.Attempt != 2 { + t.Fatalf("erwartete 2 Versuche protokolliert, hatte %d", exp.Attempt) + } +} + +// AC2 (Permanentfall): 401 vom ERP wird nicht retryed, landet sofort auf +// failed. +func TestStore_Start_PermanentFailureDoesNotRetry(t *testing.T) { + pool := openTestPool(t) + srv, _ := newReferenceERP(t, 0, true) + defer srv.Close() + + s := &Store{Pool: pool, Writer: testWriter(t, srv.URL)} + ref := Ref{TenantSlug: "tenant-c", ConnectorInstanceID: "erp-c"} + + exp, err := s.Start(context.Background(), ref, "beleg-3", []byte(`{"InvoiceNo":"RE-2026-003"}`)) + if err != nil { + t.Fatalf("Start: %v", err) + } + if exp.Status != StatusFailed { + t.Fatalf("erwartete Status failed (kein Retry bei 401), hatte %s", exp.Status) + } + if exp.RetryAt != nil { + t.Fatal("erwartete kein retry_at bei permanentem Fehler") + } +} + +// AC3 / Pruefung 3: ein bereits erfolgreich exportierter Beleg wird bei +// erneutem Start nicht ein zweites Mal an das ERP uebertragen. +func TestStore_Start_AlreadyExportedBelegNotSentTwice(t *testing.T) { + pool := openTestPool(t) + srv, received := newReferenceERP(t, 0, false) + defer srv.Close() + + s := &Store{Pool: pool, Writer: testWriter(t, srv.URL)} + ref := Ref{TenantSlug: "tenant-d", ConnectorInstanceID: "erp-d"} + payload := []byte(`{"InvoiceNo":"RE-2026-004"}`) + + first, err := s.Start(context.Background(), ref, "beleg-4", payload) + if err != nil { + t.Fatalf("erster Start: %v", err) + } + if first.Status != StatusSuccess { + t.Fatalf("erwartete Status success, hatte %s", first.Status) + } + + second, err := s.Start(context.Background(), ref, "beleg-4", payload) + if err != nil { + t.Fatalf("zweiter Start: %v", err) + } + if second.ID != first.ID || second.ERPRef != first.ERPRef { + t.Fatalf("erwartete identischen Export zurueckgegeben, hatte %+v vs %+v", first, second) + } + if len(*received) != 1 { + t.Fatalf("erwartete genau 1 Uebertragung an das ERP, hatte %d", len(*received)) + } +} + +// Pruefung 2 (ClaimDueRetries): faellige Retries werden gefunden, nicht +// faellige nicht. +func TestStore_ClaimDueRetries_OnlyReturnsDueExports(t *testing.T) { + pool := openTestPool(t) + srv, _ := newReferenceERP(t, 1, false) + defer srv.Close() + + s := &Store{Pool: pool, Writer: testWriter(t, srv.URL)} + ref := Ref{TenantSlug: "tenant-e", ConnectorInstanceID: "erp-e"} + + exp, err := s.Start(context.Background(), ref, "beleg-5", []byte(`{"InvoiceNo":"RE-2026-005"}`)) + if err != nil { + t.Fatalf("Start: %v", err) + } + if exp.Status != StatusRetryPending { + t.Fatalf("Vorbedingung: erwartete retry_pending, hatte %s", exp.Status) + } + + notYetDue, err := s.ClaimDueRetries(context.Background(), time.Now()) + if err != nil { + t.Fatalf("ClaimDueRetries (zu frueh): %v", err) + } + for _, e := range notYetDue { + if e.ID == exp.ID { + t.Fatal("Export sollte noch nicht faellig sein") + } + } + + due, err := s.ClaimDueRetries(context.Background(), exp.RetryAt.Add(time.Second)) + if err != nil { + t.Fatalf("ClaimDueRetries (faellig): %v", err) + } + found := false + for _, e := range due { + if e.ID == exp.ID { + found = true + } + } + if !found { + t.Fatal("erwartete faelligen Export in ClaimDueRetries") + } +} diff --git a/connect/migrations/0010_erp_exports.down.sql b/connect/migrations/0010_erp_exports.down.sql new file mode 100644 index 00000000..866d5e47 --- /dev/null +++ b/connect/migrations/0010_erp_exports.down.sql @@ -0,0 +1 @@ +DROP TABLE IF EXISTS erp_exports; diff --git a/connect/migrations/0010_erp_exports.up.sql b/connect/migrations/0010_erp_exports.up.sql new file mode 100644 index 00000000..6d08d2da --- /dev/null +++ b/connect/migrations/0010_erp_exports.up.sql @@ -0,0 +1,16 @@ +CREATE TABLE erp_exports ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + tenant_slug TEXT NOT NULL, + connector_instance_id TEXT NOT NULL, + beleg_id TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'pending', + attempt INT NOT NULL DEFAULT 0, + last_error TEXT, + retry_at TIMESTAMPTZ, + erp_ref TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + UNIQUE (tenant_slug, connector_instance_id, beleg_id) +); + +CREATE INDEX idx_erp_exports_retry_due ON erp_exports (retry_at) WHERE status = 'retry_pending';