144 lines
4.0 KiB
Go
144 lines
4.0 KiB
Go
package ledger
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/stretchr/testify/assert"
|
|
"github.com/stretchr/testify/require"
|
|
"github.com/tech/sendico/ledger/storage/model"
|
|
me "github.com/tech/sendico/pkg/messaging/envelope"
|
|
pmessagingreliable "github.com/tech/sendico/pkg/messaging/reliable"
|
|
"go.mongodb.org/mongo-driver/v2/bson"
|
|
"go.uber.org/zap"
|
|
)
|
|
|
|
func TestLedgerReliableProducerDispatchesLegacyOutboxRecords(t *testing.T) {
|
|
logger := zap.NewNop()
|
|
event := &model.OutboxEvent{
|
|
EventID: "entry-1",
|
|
Subject: ledgerOutboxSubject,
|
|
Payload: []byte(`{"journalEntryRef":"abc123"}`),
|
|
Attempts: 0,
|
|
}
|
|
event.SetID(bson.NewObjectID())
|
|
event.OrganizationRef = bson.NewObjectID()
|
|
|
|
store := &recordingLedgerOutboxStore{
|
|
pending: []*model.OutboxEvent{event},
|
|
}
|
|
direct := &stubDirectProducer{}
|
|
producer, _, err := newLedgerReliableProducer(logger, direct, store, nil)
|
|
require.NoError(t, err)
|
|
|
|
processed, err := producer.DispatchPending(context.Background())
|
|
require.NoError(t, err)
|
|
assert.Equal(t, 1, processed)
|
|
|
|
require.Len(t, direct.envelopes, 1)
|
|
env := direct.envelopes[0]
|
|
assert.Equal(t, outboxPublisherSender, env.GetSender())
|
|
assert.Equal(t, "ledger_outbox_sent", env.GetSignature().ToString())
|
|
|
|
var message ledgerOutboxMessage
|
|
require.NoError(t, json.Unmarshal(env.GetData(), &message))
|
|
assert.Equal(t, event.EventID, message.EventID)
|
|
assert.Equal(t, event.Subject, message.Subject)
|
|
assert.Equal(t, event.OrganizationRef.Hex(), message.OrganizationRef)
|
|
assert.JSONEq(t, `{"journalEntryRef":"abc123"}`, string(message.Payload))
|
|
|
|
require.Len(t, store.markedSent, 1)
|
|
assert.Equal(t, *event.GetID(), store.markedSent[0])
|
|
assert.Empty(t, store.markedFailed)
|
|
assert.Empty(t, store.incremented)
|
|
}
|
|
|
|
func TestLedgerReliableProducerMarksFailedOnDispatchError(t *testing.T) {
|
|
logger := zap.NewNop()
|
|
event := &model.OutboxEvent{
|
|
EventID: "entry-2",
|
|
Subject: ledgerOutboxSubject,
|
|
Payload: []byte(`{"journalEntryRef":"xyz789"}`),
|
|
Attempts: pmessagingreliable.DefaultSettings().MaxAttempts - 1,
|
|
}
|
|
event.SetID(bson.NewObjectID())
|
|
event.OrganizationRef = bson.NewObjectID()
|
|
|
|
store := &recordingLedgerOutboxStore{
|
|
pending: []*model.OutboxEvent{event},
|
|
}
|
|
direct := &stubDirectProducer{err: errors.New("publish failed")}
|
|
producer, _, err := newLedgerReliableProducer(logger, direct, store, nil)
|
|
require.NoError(t, err)
|
|
|
|
processed, err := producer.DispatchPending(context.Background())
|
|
require.NoError(t, err)
|
|
assert.Equal(t, 1, processed)
|
|
|
|
require.Len(t, store.incremented, 1)
|
|
assert.Equal(t, *event.GetID(), store.incremented[0])
|
|
require.Len(t, store.markedFailed, 1)
|
|
assert.Equal(t, *event.GetID(), store.markedFailed[0])
|
|
assert.Empty(t, store.markedSent)
|
|
}
|
|
|
|
type recordingLedgerOutboxStore struct {
|
|
mu sync.Mutex
|
|
|
|
pending []*model.OutboxEvent
|
|
|
|
markedSent []bson.ObjectID
|
|
markedFailed []bson.ObjectID
|
|
incremented []bson.ObjectID
|
|
}
|
|
|
|
func (s *recordingLedgerOutboxStore) Create(context.Context, *model.OutboxEvent) error {
|
|
return nil
|
|
}
|
|
|
|
func (s *recordingLedgerOutboxStore) ListPending(context.Context, int) ([]*model.OutboxEvent, error) {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
events := s.pending
|
|
s.pending = nil
|
|
return events, nil
|
|
}
|
|
|
|
func (s *recordingLedgerOutboxStore) MarkSent(_ context.Context, eventRef bson.ObjectID, _ time.Time) error {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
s.markedSent = append(s.markedSent, eventRef)
|
|
return nil
|
|
}
|
|
|
|
func (s *recordingLedgerOutboxStore) MarkFailed(_ context.Context, eventRef bson.ObjectID) error {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
s.markedFailed = append(s.markedFailed, eventRef)
|
|
return nil
|
|
}
|
|
|
|
func (s *recordingLedgerOutboxStore) IncrementAttempts(_ context.Context, eventRef bson.ObjectID) error {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
s.incremented = append(s.incremented, eventRef)
|
|
return nil
|
|
}
|
|
|
|
type stubDirectProducer struct {
|
|
mu sync.Mutex
|
|
envelopes []me.Envelope
|
|
err error
|
|
}
|
|
|
|
func (p *stubDirectProducer) SendMessage(env me.Envelope) error {
|
|
p.mu.Lock()
|
|
defer p.mu.Unlock()
|
|
p.envelopes = append(p.envelopes, env)
|
|
return p.err
|
|
}
|