ProofedMesh/services/attest-reconciler/internal/reconcile/reconciler.go

// 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()
	}
}