56 lines
2.1 KiB
Go
56 lines
2.1 KiB
Go
package notifications
|
|
|
|
import (
|
|
"context"
|
|
|
|
me "github.com/tech/sendico/pkg/messaging/envelope"
|
|
gmessaging "github.com/tech/sendico/pkg/generated/gmessaging"
|
|
moh "github.com/tech/sendico/pkg/messaging/notifications/object/handler"
|
|
np "github.com/tech/sendico/pkg/messaging/notifications/processor"
|
|
"github.com/tech/sendico/pkg/mlogger"
|
|
"github.com/tech/sendico/pkg/model"
|
|
nm "github.com/tech/sendico/pkg/model/notification"
|
|
"github.com/tech/sendico/pkg/mservice"
|
|
"go.mongodb.org/mongo-driver/bson/primitive"
|
|
"go.uber.org/zap"
|
|
"google.golang.org/protobuf/proto"
|
|
)
|
|
|
|
type ObjectNotificaionProcessor struct {
|
|
logger mlogger.Logger
|
|
handler moh.ObjectUpdateHandler
|
|
event model.NotificationEvent
|
|
}
|
|
|
|
func (ounp *ObjectNotificaionProcessor) Process(ctx context.Context, envelope me.Envelope) error {
|
|
var msg gmessaging.ObjectUpdatedEvent
|
|
if err := proto.Unmarshal(envelope.GetData(), &msg); err != nil {
|
|
ounp.logger.Warn("Failed to unmarshall envelope", zap.Error(err), zap.String("topic", ounp.event.ToString()))
|
|
return err
|
|
}
|
|
actorAccountRef, err := primitive.ObjectIDFromHex(msg.ActorAccountRef)
|
|
if err != nil {
|
|
ounp.logger.Warn("Failed to restore actor account reference", zap.Error(err), zap.String("topic", ounp.event.ToString()), zap.String("actor_account_ref", msg.ActorAccountRef))
|
|
return err
|
|
}
|
|
objectRef, err := primitive.ObjectIDFromHex(msg.ObjectRef)
|
|
if err != nil {
|
|
ounp.logger.Warn("Failed to restore object reference", zap.Error(err), zap.String("topic", ounp.event.ToString()), zap.String("object_ref", msg.ObjectRef))
|
|
return err
|
|
}
|
|
|
|
return ounp.handler(ctx, envelope.GetSignature().GetType(), objectRef, actorAccountRef, envelope.GetSignature().GetAction())
|
|
}
|
|
|
|
func (acnp *ObjectNotificaionProcessor) GetSubject() model.NotificationEvent {
|
|
return acnp.event
|
|
}
|
|
|
|
func NewObjectChangeMessageProcessor(logger mlogger.Logger, handler moh.ObjectUpdateHandler, objectType mservice.Type, action nm.NotificationAction) np.EnvelopeProcessor {
|
|
return &ObjectNotificaionProcessor{
|
|
logger: logger.Named("message_processor"),
|
|
handler: handler,
|
|
event: NewObjectNotification(objectType, action),
|
|
}
|
|
}
|