billing/adapter/validationissue.go

package billingadapter

import (
	"context"
	"crypto/sha256"
	"time"

	"entgo.io/ent/dialect/sql"
	"github.com/samber/lo"

	"github.com/openmeterio/openmeter/openmeter/billing"
	"github.com/openmeterio/openmeter/openmeter/ent/db"
	"github.com/openmeterio/openmeter/openmeter/ent/db/billinginvoicevalidationissue"
	"github.com/openmeterio/openmeter/pkg/clock"
)

type validationIssueWithDedupe struct {
	issue billing.ValidationIssue
	hash  []byte
}

func issueDedupeHash(issue billing.ValidationIssue) []byte {
	algo := sha256.New()

	algo.Write([]byte(issue.Severity))
	algo.Write([]byte(issue.Code))
	algo.Write([]byte(issue.Message))
	algo.Write([]byte(issue.Component))
	algo.Write([]byte(issue.Path))
	return algo.Sum(nil)
}

// persistValidationIssues persists the validation issues for the given invoice, it will remove any
// existing issues that are not present in the new list. It relies on consistent hashing to deduplicate
// issues.
func (a *adapter) persistValidationIssues(ctx context.Context, invoice billing.InvoiceID, issues []billing.ValidationIssue) error {
	// FIXME (pmarton): Why do we need to deduplicate issues?
	hashedIssues := lo.FindUniquesBy(
		lo.Map(issues, func(issue billing.ValidationIssue, _ int) validationIssueWithDedupe {
			return validationIssueWithDedupe{
				issue: issue,
				hash:  issueDedupeHash(issue),
			}
		}),
		func(issue validationIssueWithDedupe) string {
			return string(issue.hash)
		},
	)

	err := a.db.BillingInvoiceValidationIssue.Update().
		Where(billinginvoicevalidationissue.InvoiceID(invoice.ID)).
		Where(billinginvoicevalidationissue.Namespace(invoice.Namespace)).
		Where(billinginvoicevalidationissue.DedupeHashNotIn(
			lo.Map(hashedIssues, func(hashedIssue validationIssueWithDedupe, _ int) []byte {
				return hashedIssue.hash
			})...)).
		Where(billinginvoicevalidationissue.DeletedAtIsNil()).
		SetDeletedAt(clock.Now()).
		Exec(ctx)
	if err != nil {
		return err
	}

	return a.db.BillingInvoiceValidationIssue.MapCreateBulk(hashedIssues, func(c *db.BillingInvoiceValidationIssueCreate, i int) {
		hash := hashedIssues[i].hash
		issue := hashedIssues[i].issue

		c.SetNamespace(invoice.Namespace).
			SetInvoiceID(invoice.ID).
			SetSeverity(issue.Severity).
			SetMessage(issue.Message).
			SetComponent(string(issue.Component)).
			SetDedupeHash(hash)
		if issue.Code != "" {
			c.SetCode(issue.Code)
		}

		if issue.Path != "" {
			c.SetPath(issue.Path)
		}
	}).OnConflict(
		sql.ConflictColumns(
			billinginvoicevalidationissue.FieldNamespace,
			billinginvoicevalidationissue.FieldInvoiceID,
			billinginvoicevalidationissue.FieldDedupeHash,
		),
	).
		UpdateNewValues().
		Update(func(u *db.BillingInvoiceValidationIssueUpsert) {
			u.ClearDeletedAt()
			u.SetUpdatedAt(clock.Now())
		}).Exec(ctx)
}

type ValidationIssueWithDBMeta struct {
	billing.ValidationIssue

	ID        string     `json:"id"`
	DeletedAt *time.Time `json:"deletedAt,omitempty"`
}

// IntropectValidationIssues returns the validation issues for the given invoice, this is not
// exposed via the adpter interface, as it's only used by tests to validate the state of the
// database.
func (a *adapter) IntrospectValidationIssues(ctx context.Context, invoice billing.InvoiceID) ([]ValidationIssueWithDBMeta, error) {
	issues, err := a.db.BillingInvoiceValidationIssue.Query().
		Where(billinginvoicevalidationissue.InvoiceID(invoice.ID)).
		Where(billinginvoicevalidationissue.Namespace(invoice.Namespace)).
		Order(db.Asc(billinginvoicevalidationissue.FieldCreatedAt)).
		All(ctx)
	if err != nil {
		return nil, err
	}

	return lo.Map(issues, func(issue *db.BillingInvoiceValidationIssue, _ int) ValidationIssueWithDBMeta {
		return ValidationIssueWithDBMeta{
			ValidationIssue: billing.ValidationIssue{
				Severity:  issue.Severity,
				Message:   issue.Message,
				Code:      lo.FromPtr(issue.Code),
				Component: billing.ComponentName(issue.Component),
				Path:      lo.FromPtr(issue.Path),
			},
			ID:        issue.ID,
			DeletedAt: issue.DeletedAt,
		}
	}), nil
}