package reconcile import ( "context" "errors" "log/slog" "os" "sync" "testing" "time" "basicstack.de/proofedmesh/services/attest-reconciler/internal/attest" "basicstack.de/proofedmesh/services/attest-reconciler/internal/metrics" "basicstack.de/proofedmesh/services/attest-reconciler/internal/store" ) // fakeStore models per_diem_rates in-memory. ListPending honours the // WHERE rate_change_attestation_ref IS NULL contract โ€” any row that already // has a ref is filtered out. AttachAttestation sets the ref and is the only // way a row leaves the pending set. type fakeStore struct { mu sync.Mutex rows []storedRow } type storedRow struct { row store.PendingRow ref string // "" = no attestation yet } func (f *fakeStore) seed(row store.PendingRow, ref string) { f.mu.Lock() defer f.mu.Unlock() f.rows = append(f.rows, storedRow{row: row, ref: ref}) } func (f *fakeStore) ListPending(_ context.Context, limit int) ([]store.PendingRow, error) { f.mu.Lock() defer f.mu.Unlock() out := make([]store.PendingRow, 0, len(f.rows)) for _, r := range f.rows { if r.ref != "" { continue } out = append(out, r.row) if limit > 0 && len(out) >= limit { break } } return out, nil } func (f *fakeStore) AttachAttestation(_ context.Context, rateID, ref string) error { f.mu.Lock() defer f.mu.Unlock() for i := range f.rows { if f.rows[i].row.Rate.RateID == rateID { if f.rows[i].ref != "" { // Already attested; treat as idempotent no-op in the fake โ€” // production AttachAttestation uses WHERE ref IS NULL and // returns ErrNotFound for a conflict, which the reconciler // surfaces as a failure. Our test paths never exercise that. return nil } f.rows[i].ref = ref return nil } } return store.ErrNotFound } func (f *fakeStore) refFor(rateID string) string { f.mu.Lock() defer f.mu.Unlock() for _, r := range f.rows { if r.row.Rate.RateID == rateID { return r.ref } } return "" } // fakeSigner returns a deterministic envelope + content_hash per rate_id so // resubmits land on the same TLog idempotency key. type fakeSigner struct { calls int mu sync.Mutex } func (f *fakeSigner) SignDSSE(_ context.Context, keyRef string, stmt attest.Statement) (attest.Envelope, string, error) { f.mu.Lock() f.calls++ f.mu.Unlock() // The statement's subject.name is "per_diem_rate/". rateID := "" if len(stmt.Subject) > 0 { rateID = stmt.Subject[0].Name } env := attest.Envelope{ PayloadType: "application/vnd.in-toto+json", Payload: "stub-payload-" + rateID, Signatures: []attest.EnvelopeSignature{{KeyID: keyRef, Sig: "stub-sig-" + rateID}}, } hash, err := attest.EnvelopeContentHash(env) return env, hash, err } // fakeSubmitter remembers which content_hashes it saw. The TLog contract says // a resubmit returns the SAME receipt โ€” the fake enforces that. type fakeSubmitter struct { mu sync.Mutex seen map[string]attest.LogReceipt failWith error } func (f *fakeSubmitter) Submit(_ context.Context, env attest.Envelope, contentHash string) (attest.LogReceipt, error) { if f.failWith != nil { return attest.LogReceipt{}, f.failWith } f.mu.Lock() defer f.mu.Unlock() if f.seen == nil { f.seen = map[string]attest.LogReceipt{} } if existing, ok := f.seen[contentHash]; ok { return existing, nil } rcpt := attest.LogReceipt{ LogIndex: int64(len(f.seen) + 1), ContentHash: contentHash, LoggedAt: time.Now().UTC(), } f.seen[contentHash] = rcpt return rcpt, nil } func (f *fakeSubmitter) calls() int { f.mu.Lock() defer f.mu.Unlock() return len(f.seen) } func pendingRow(id string, kind attest.ChangeKind) store.PendingRow { return store.PendingRow{ Rate: attest.Rate{ RateID: id, Country: "DE", Currency: "EUR", FullDay: "28.00", PartialDayMin8h: "14.00", ArrivalDay: "14.00", DepartureDay: "14.00", ValidFrom: time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC), SourceRegulation: "BRKG ยง 9 Abs. 1", CreatedBy: "stub-admin", CreatedAt: time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC), }, Kind: kind, } } func newReconciler(st store.PendingRateStore, signer attest.Signer, sub attest.Submitter) (*Reconciler, *metrics.Registry) { reg := metrics.New() return &Reconciler{ Store: st, Orchestrator: attest.Orchestrator{ Signer: signer, Submitter: sub, KeyRef: "trust-api/rate-change-signer-2026-q4", }, Metrics: reg, Logger: slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelWarn})), Issuer: "https://auth.basicstack.de", Subject: "attest-reconciler@proofedmesh", SessionID: "test", Now: func() time.Time { return time.Date(2026, 10, 25, 12, 0, 0, 0, time.UTC) }, }, reg } // TestReconciler_SignsPendingRows is the happy path: two pending rows get // signed, logged, and attached in one tick. func TestReconciler_SignsPendingRows(t *testing.T) { st := &fakeStore{} st.seed(pendingRow("PDR-1", attest.ChangeCreate), "") st.seed(pendingRow("PDR-2", attest.ChangeCreate), "") signer := &fakeSigner{} sub := &fakeSubmitter{} r, reg := newReconciler(st, signer, sub) rep, err := r.RunOnce(context.Background()) if err != nil { t.Fatalf("RunOnce err: %v", err) } if rep.DirtyRows != 2 || rep.Attempts != 2 || rep.Successes != 2 || rep.Failures != 0 { t.Fatalf("unexpected report: %+v", rep) } if ref := st.refFor("PDR-1"); ref == "" { t.Errorf("PDR-1 has no attestation ref after RunOnce") } if ref := st.refFor("PDR-2"); ref == "" { t.Errorf("PDR-2 has no attestation ref after RunOnce") } m := reg.Snapshot() if m.DirtyRows != 2 || m.ReconciliationAttempts != 2 || m.ReconciliationFailures != 0 { t.Errorf("unexpected metrics snapshot: %+v", m) } } // TestReconciler_SkipsAlreadyAttestedRows is the spec's required invariant: // bereits attestierte Rows werden nicht erneut signiert. We seed a store with // one already-signed row and one pending row and assert that the signer was // called exactly once (for the pending row only). func TestReconciler_SkipsAlreadyAttestedRows(t *testing.T) { st := &fakeStore{} st.seed(pendingRow("PDR-already-signed", attest.ChangeCreate), "sha256:cafebabe"+repeat("00", 28)) st.seed(pendingRow("PDR-still-pending", attest.ChangeCreate), "") signer := &fakeSigner{} sub := &fakeSubmitter{} r, _ := newReconciler(st, signer, sub) rep, err := r.RunOnce(context.Background()) if err != nil { t.Fatalf("RunOnce err: %v", err) } if rep.DirtyRows != 1 { t.Fatalf("expected 1 dirty row, got %d (full report: %+v)", rep.DirtyRows, rep) } if rep.Attempts != 1 { t.Errorf("expected 1 attempt (only the pending row), got %d", rep.Attempts) } if signer.calls != 1 { t.Errorf("expected signer to be called once, got %d calls", signer.calls) } if ref := st.refFor("PDR-already-signed"); ref != "sha256:cafebabe"+repeat("00", 28) { t.Errorf("already-signed row's ref changed: %q", ref) } if ref := st.refFor("PDR-still-pending"); ref == "" { t.Errorf("still-pending row has no attestation ref after RunOnce") } } // TestReconciler_TLogIdempotencyOnReRun shows the Transparency Log's // idempotency guarantee: if the ref somehow fails to persist (e.g. the DB // hiccup after a successful log entry), the next tick re-signs the row and // the TLog returns the SAME log_index. This proves kein Doppelsignieren. func TestReconciler_TLogIdempotencyOnReRun(t *testing.T) { st := &fakeStore{} st.seed(pendingRow("PDR-1", attest.ChangeCreate), "") signer := &fakeSigner{} sub := &fakeSubmitter{} r, _ := newReconciler(st, signer, sub) // Tick 1: sign and log; the fake store attaches the ref. rep1, err := r.RunOnce(context.Background()) if err != nil || rep1.Successes != 1 { t.Fatalf("tick 1 failed: %+v err=%v", rep1, err) } ref1 := st.refFor("PDR-1") // Simulate the "ref persist lost" case by clearing the ref so the row is // pending again. In production that would require a different root cause // (DB rollback, async worker crash), but it is the exact scenario the // reconciler is here to recover from. st.mu.Lock() for i := range st.rows { if st.rows[i].row.Rate.RateID == "PDR-1" { st.rows[i].ref = "" } } st.mu.Unlock() // Tick 2: must re-sign the same row AND land on the same TLog entry. rep2, err := r.RunOnce(context.Background()) if err != nil || rep2.Successes != 1 { t.Fatalf("tick 2 failed: %+v err=%v", rep2, err) } ref2 := st.refFor("PDR-1") if ref1 != ref2 { t.Errorf("content_hash changed between ticks: %q vs %q (TLog idempotency broken)", ref1, ref2) } if got := sub.calls(); got != 1 { t.Errorf("TLog should hold exactly one entry across two ticks, got %d", got) } } // TestReconciler_RecordsFailuresWhenTLogDown ensures the metrics + // retry-next-tick semantics are correct when the Transparency Log is // unreachable; the row must stay pending. func TestReconciler_RecordsFailuresWhenTLogDown(t *testing.T) { st := &fakeStore{} st.seed(pendingRow("PDR-1", attest.ChangeCreate), "") signer := &fakeSigner{} sub := &fakeSubmitter{failWith: errors.New("tlog down")} r, reg := newReconciler(st, signer, sub) rep, err := r.RunOnce(context.Background()) if err != nil { t.Fatalf("RunOnce err: %v", err) } if rep.Successes != 0 || rep.Failures != 1 { t.Fatalf("expected 0 successes / 1 failure, got %+v", rep) } if ref := st.refFor("PDR-1"); ref != "" { t.Errorf("row acquired a ref despite TLog failure: %q", ref) } if m := reg.Snapshot(); m.ReconciliationFailures != 1 { t.Errorf("failure metric not incremented: %+v", m) } } func TestReconciler_ContextCancelStopsLoop(t *testing.T) { st := &fakeStore{} for i := 0; i < 10; i++ { st.seed(pendingRow("PDR-"+itoa(i), attest.ChangeCreate), "") } signer := &fakeSigner{} sub := &fakeSubmitter{} r, _ := newReconciler(st, signer, sub) ctx, cancel := context.WithCancel(context.Background()) cancel() // cancel before RunOnce starts the per-row loop _, err := r.RunOnce(ctx) if err == nil { t.Fatal("expected ctx.Err() from RunOnce, got nil") } } // --- small helpers to keep the test file free of fmt/strings deps --- func repeat(s string, n int) string { out := "" for i := 0; i < n; i++ { out += s } return out } func itoa(n int) string { if n == 0 { return "0" } neg := n < 0 if neg { n = -n } var buf [20]byte i := len(buf) for n > 0 { i-- buf[i] = byte('0' + n%10) n /= 10 } if neg { i-- buf[i] = '-' } return string(buf[i:]) }