audit-go/audit/api/api_routable.go

package api

import (
	"context"
	"errors"
	"fmt"
	"github.com/google/uuid"
	"google.golang.org/protobuf/proto"

	"dev.azure.com/schwarzit/schwarzit.stackit-core-platform/common-audit.git/audit/messaging"
	auditV1 "dev.azure.com/schwarzit/schwarzit.stackit-core-platform/common-audit.git/gen/go/audit/v1"
)

// routableTopicNameResolver implements TopicNameResolver.
// Resolves topic names by concatenating topic type prefixes with routing identifiers.
type routableTopicNameResolver struct {
	organizationTopicPrefix string
	ProjectTopicPrefix      string
	// If no identifier is provided for routing, it will be routed to a system topic
	systemTopicName string
}

// Resolve implements TopicNameResolver.Resolve.
func (r *routableTopicNameResolver) Resolve(routingIdentifier *RoutingIdentifier) (string, error) {

	if routingIdentifier == nil {
		return r.systemTopicName, nil
	}

	switch routingIdentifier.Type {
	case RoutingIdentifierTypeOrganization:
		return fmt.Sprintf("topic://%s/%s", r.organizationTopicPrefix, routingIdentifier.Identifier), nil
	case RoutingIdentifierTypeProject:
		return fmt.Sprintf("topic://%s/%s", r.ProjectTopicPrefix, routingIdentifier.Identifier), nil
	default:
		return "", ErrUnsupportedObjectIdentifierType
	}
}

// topicNameConfig provides topic name information required for the topic name resolution.
type topicNameConfig struct {
	OrganizationTopicPrefix string
	ProjectTopicPrefix      string
	SystemTopicName         string
}

// routableAuditApi is an implementation of AuditApi.
// Warning: It is only there for local (compatibility) testing.
// DO NOT USE IT!
type routableAuditApi struct {
	messagingApi      *messaging.MessagingApi
	topicNameResolver *TopicNameResolver
	validator         *ProtobufValidator
}

// NewRoutableAuditApi can be used to initialize the audit log api.
func newRoutableAuditApi(
	messagingApi *messaging.MessagingApi,
	topicNameConfig topicNameConfig,
	validator ProtobufValidator,
) (*AuditApi, error) {

	if messagingApi == nil {
		return nil, errors.New("messaging api nil")
	}

	// Topic resolver
	var topicNameResolver TopicNameResolver = &routableTopicNameResolver{
		organizationTopicPrefix: topicNameConfig.OrganizationTopicPrefix,
		ProjectTopicPrefix:      topicNameConfig.ProjectTopicPrefix,
		systemTopicName:         topicNameConfig.SystemTopicName,
	}

	// Audit api
	var auditApi AuditApi = &routableAuditApi{
		messagingApi:      messagingApi,
		topicNameResolver: &topicNameResolver,
		validator:         &validator,
	}

	return &auditApi, nil
}

// Log implements AuditApi.Log
func (a *routableAuditApi) Log(
	ctx context.Context,
	event *auditV1.AuditEvent,
	visibility auditV1.Visibility,
	routingIdentifier *RoutingIdentifier,
	objectIdentifier *auditV1.ObjectIdentifier,
) error {

	cloudEvent, err := a.ValidateAndSerialize(event, visibility, routingIdentifier, objectIdentifier)
	if err != nil {
		return err
	}

	return a.Send(ctx, routingIdentifier, cloudEvent)
}

// ValidateAndSerialize implements AuditApi.ValidateAndSerialize
func (a *routableAuditApi) ValidateAndSerialize(
	event *auditV1.AuditEvent,
	visibility auditV1.Visibility,
	routingIdentifier *RoutingIdentifier,
	objectIdentifier *auditV1.ObjectIdentifier,
) (*CloudEvent, error) {

	routableEvent, err := validateAndSerializePartially(
		a.validator,
		event,
		visibility,
		routingIdentifier,
		objectIdentifier)

	if err != nil {
		return nil, err
	}

	routableEventBytes, err := proto.Marshal(routableEvent)
	if err != nil {
		return nil, err
	}

	message := CloudEvent{
		specVersion:     "1.0",
		source:          event.EventSource,
		id:              uuid.NewString(),
		time:            event.EventTimeStamp.AsTime(),
		dataContentType: "application/cloudevents+protobuf",
		dataType:        fmt.Sprintf("%v", routableEvent.ProtoReflect().Descriptor().FullName()),
		data:            routableEventBytes,
	}

	return &message, nil
}

// Send implements AuditApi.Send
func (a *routableAuditApi) Send(
	ctx context.Context,
	routingIdentifier *RoutingIdentifier,
	cloudEvent *CloudEvent,
) error {

	return send(a.topicNameResolver, a.messagingApi, ctx, routingIdentifier, cloudEvent)
}