Monitoring

package observability

import (
	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/push"
)

// InvoicesGeneratedTotal, InvoiceGenerationDuration, InvoicesOverdue, and
// LedgerStatusDrift are registered into a dedicated registry (not the
// default global one) so PushBatchMetrics can push exactly this set and
// nothing accidentally pulled in from an unrelated package's init().
var (
	Registry = prometheus.NewRegistry()

	InvoicesGeneratedTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
		Name: "invoices_generated_total",
		Help: "Invoices generated by a billing run, labeled by outcome.",
	}, []string{"outcome"}) // outcome: created | skipped_existing | failed

	InvoiceGenerationDuration = prometheus.NewHistogram(prometheus.HistogramOpts{
		Name:    "invoice_generation_duration_seconds",
		Help:    "Time to generate and persist a single flat's invoice.",
		Buckets: prometheus.DefBuckets,
	})

	InvoicesOverdue = prometheus.NewGaugeVec(prometheus.GaugeOpts{
		Name: "invoices_overdue",
		Help: "Current count of overdue invoices, labeled by tenant.",
	}, []string{"tenant_id"})

	LedgerStatusDrift = prometheus.NewCounterVec(prometheus.CounterOpts{
		Name: "ledger_status_drift_total",
		Help: "Invoices whose stored status did not match the status re-derived from payment_records.",
	}, []string{"tenant_id"})
)

func init() {
	Registry.MustRegister(InvoicesGeneratedTotal, InvoiceGenerationDuration, InvoicesOverdue, LedgerStatusDrift)
}

// PushBatchMetrics pushes the current values of every metric in Registry
// to the given Pushgateway URL under a fixed job name. It is called once,
// right before cmd/billing-runner and cmd/reconciler exit — there is no
// ongoing scrape loop to rely on, so this single push is the job's only
// chance to report anything.
func PushBatchMetrics(pushgatewayURL, jobName string) error {
	return push.New(pushgatewayURL, jobName).Gatherer(Registry).Push()
}

package service

import (
	"context"
	"time"

	"github.com/google/uuid"

	"github.com/hoa-platform/backend/services/maintenance-service/internal/domain"
	"github.com/hoa-platform/backend/services/maintenance-service/internal/observability"
)

// recordOutcome wraps CreateIfAbsent's result and any error into the
// outcome label observability.InvoicesGeneratedTotal expects, and times
// the whole per-flat attempt. It replaces the direct call to
// r.invoices.CreateIfAbsent inside processFlat.
func (r *BillingRunner) instrumentedCreateIfAbsent(ctx context.Context, invoice domain.Invoice, lineItems []domain.InvoiceLineItem) (bool, error) {
	start := time.Now()
	created, err := r.invoices.CreateIfAbsent(ctx, invoice, lineItems)
	observability.InvoiceGenerationDuration.Observe(time.Since(start).Seconds())

	switch {
	case err != nil:
		observability.InvoicesGeneratedTotal.WithLabelValues("failed").Inc()
	case created:
		observability.InvoicesGeneratedTotal.WithLabelValues("created").Inc()
	default:
		observability.InvoicesGeneratedTotal.WithLabelValues("skipped_existing").Inc()
	}
	return created, err
}

var _ = uuid.Nil

package postgres

import (
	"context"

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

	"github.com/hoa-platform/backend/services/maintenance-service/internal/domain"
)

type ReconciliationRepository struct {
	pool *pgxpool.Pool
}

func NewReconciliationRepository(pool *pgxpool.Pool) *ReconciliationRepository {
	return &ReconciliationRepository{pool: pool}
}

type LedgerRow struct {
	InvoiceID     uuid.UUID
	TenantID      uuid.UUID
	StoredStatus  domain.InvoiceStatus
	TotalPaise    domain.Paise
	PaidPaise     domain.Paise
	IsPastDueDate bool
}

// ListForReconciliation returns every non-terminal invoice (not yet paid
// or written off) along with the raw numbers needed to re-derive its
// status independently of whatever is currently stored.
func (r *ReconciliationRepository) ListForReconciliation(ctx context.Context) ([]LedgerRow, error) {
	const q = `
		SELECT id, tenant_id, status, total_paise, paid_paise,
		       (billing_period + INTERVAL '30 days') < now() AS is_past_due_date
		FROM invoices
		WHERE status NOT IN ('paid', 'written_off')`

	rows, err := r.pool.Query(ctx, q)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var out []LedgerRow
	for rows.Next() {
		var lr LedgerRow
		if err := rows.Scan(&lr.InvoiceID, &lr.TenantID, &lr.StoredStatus, &lr.TotalPaise, &lr.PaidPaise, &lr.IsPastDueDate); err != nil {
			return nil, err
		}
		out = append(out, lr)
	}
	return out, rows.Err()
}

// PromoteToOverdue transitions one invoice to overdue via a direct
// UPDATE. Doing this with raw SQL rather than routing back through
// PaymentRepository (which only knows how to apply a payment or a
// write-off) is flagged explicitly in this lesson's quiz as a spot where
// the state-machine rule and this query could drift apart if
// domain.legalTransitions ever changes without this query being revisited.
func (r *ReconciliationRepository) PromoteToOverdue(ctx context.Context, invoiceID uuid.UUID) error {
	_, err := r.pool.Exec(ctx, `UPDATE invoices SET status = 'overdue' WHERE id = $1`, invoiceID)
	return err
}

package service

import (
	"context"
	"log"

	"github.com/google/uuid"

	"github.com/hoa-platform/backend/services/maintenance-service/internal/domain"
	"github.com/hoa-platform/backend/services/maintenance-service/internal/observability"
	"github.com/hoa-platform/backend/services/maintenance-service/internal/repository/postgres"
)

type Reconciler struct {
	repo *postgres.ReconciliationRepository
}

func NewReconciler(repo *postgres.ReconciliationRepository) *Reconciler {
	return &Reconciler{repo: repo}
}

// Run re-derives what each non-terminal invoice's status ought to be from
// its raw numbers, using the exact same domain.CanTransition rules the
// payment path is bound by, and compares that to what is actually stored.
// A mismatch increments LedgerStatusDrift and is logged for investigation
// — it is never silently auto-corrected, because a drift is evidence of a
// bug somewhere else, not routine housekeeping.
func (rec *Reconciler) Run(ctx context.Context) error {
	rows, err := rec.repo.ListForReconciliation(ctx)
	if err != nil {
		return err
	}

	overdueByTenant := map[uuid.UUID]int{}

	for _, row := range rows {
		derived := deriveStatus(row)

		if derived != row.StoredStatus {
			observability.LedgerStatusDrift.WithLabelValues(row.TenantID.String()).Inc()
			log.Printf("ledger drift: invoice %s tenant %s stored=%s derived=%s",
				row.InvoiceID, row.TenantID, row.StoredStatus, derived)

			if derived == domain.InvoiceStatusOverdue && domain.CanTransition(row.StoredStatus, domain.InvoiceStatusOverdue) {
				if err := rec.repo.PromoteToOverdue(ctx, row.InvoiceID); err != nil {
					log.Printf("failed to promote invoice %s to overdue: %v", row.InvoiceID, err)
				}
			}
		}

		if derived == domain.InvoiceStatusOverdue {
			overdueByTenant[row.TenantID]++
		}
	}

	for tenantID, count := range overdueByTenant {
		observability.InvoicesOverdue.WithLabelValues(tenantID.String()).Set(float64(count))
	}
	return nil
}

func deriveStatus(row postgres.LedgerRow) domain.InvoiceStatus {
	switch {
	case row.PaidPaise >= row.TotalPaise:
		return domain.InvoiceStatusPaid
	case row.IsPastDueDate:
		return domain.InvoiceStatusOverdue
	case row.PaidPaise > 0:
		return domain.InvoiceStatusPartiallyPaid
	default:
		return domain.InvoiceStatusIssued
	}
}

reconciler := service.NewReconciler(postgres.NewReconciliationRepository(pool))
if err := reconciler.Run(context.Background()); err != nil {
	log.Fatalf("reconciliation run failed: %v", err)
}

if err := observability.PushBatchMetrics(os.Getenv("PUSHGATEWAY_URL"), "maintenance_reconciler"); err != nil {
	log.Printf("failed to push metrics: %v", err)
}
log.Println("reconciliation run complete")