// Package reconcile re-signs per_diem_rates rows that are missing an // attestation reference. Depends only on the Orchestrator and PendingRateStore // interfaces so tests can swap in fakes with zero mocking framework. package reconcile import ( "context" "errors" "fmt" "log/slog" "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" ) // Reconciler re-signs pending rows one tick at a time. The scheduler owns the // timer; RunOnce is the unit of work. type Reconciler struct { Store store.PendingRateStore Orchestrator attest.Orchestrator Metrics *metrics.Registry Logger *slog.Logger Issuer string // OIDC issuer stamped on the reconciler's own attestations Subject string // service identity, e.g. "attest-reconciler@proofedmesh" SessionID string // deterministic marker for provenance; empty is fine BatchLimit int // max rows per tick; 0 => 100 Now func() time.Time } // RunOnce executes a single reconciliation tick: fetches pending rows, // re-signs each, and persists the attestation ref. Returns a Report that // callers (and tests) can inspect without reading metrics. func (r *Reconciler) RunOnce(ctx context.Context) (Report, error) { if r.Store == nil { return Report{}, errors.New("reconcile.Reconciler: Store is nil") } if r.Now == nil { r.Now = time.Now } limit := r.BatchLimit if limit <= 0 { limit = 100 } rows, err := r.Store.ListPending(ctx, limit) if err != nil { return Report{}, fmt.Errorf("list pending: %w", err) } rep := Report{DirtyRows: len(rows)} if r.Metrics != nil { r.Metrics.SetDirtyRows(int64(len(rows))) } for _, row := range rows { if err := ctx.Err(); err != nil { return rep, err } rep.Attempts++ if r.Metrics != nil { r.Metrics.IncAttempts() } ev := attest.RateChangeEvent{ Kind: row.Kind, Rate: row.Rate, PerformedBy: attest.PerformedBy{ Issuer: r.Issuer, Subject: r.Subject, SessionID: r.SessionID, }, PerformedAt: r.Now().UTC(), } contentHash, _, err := r.Orchestrator.SignAndLog(ctx, ev) if err != nil { rep.Failures++ if r.Metrics != nil { r.Metrics.IncFailures() } if r.Logger != nil { r.Logger.Warn("reconcile: sign+log failed; will retry next tick", "rate_id", row.Rate.RateID, "err", err) } continue } if err := r.Store.AttachAttestation(ctx, row.Rate.RateID, contentHash); err != nil { rep.Failures++ if r.Metrics != nil { r.Metrics.IncFailures() } if r.Logger != nil { r.Logger.Error("reconcile: attach attestation failed after successful sign+log; row stays pending but TLog entry exists", "rate_id", row.Rate.RateID, "content_hash", contentHash, "err", err) } continue } rep.Successes++ if r.Logger != nil { r.Logger.Info("reconcile: row attested", "rate_id", row.Rate.RateID, "kind", string(row.Kind), "content_hash", contentHash) } } return rep, nil } // Report captures what a single tick did; the scheduler logs it per tick and // tests assert on it. type Report struct { DirtyRows int Attempts int Successes int Failures int } // NoteIdempotentNoop lets higher layers record that the TLog returned the // same log_index for a resubmit (proves idempotency guarantee); the signing // path itself cannot know that from a single call because the server treats // 201 and 200 as both "ok". func (r *Reconciler) NoteIdempotentNoop() { if r.Metrics != nil { r.Metrics.IncIdempotentNoopHits() } }