payments/internal/storage/connectors_test.go

package storage

import (
	"context"
	"encoding/json"
	"fmt"
	"testing"

	"github.com/hanzo-fi/go-libs/v5/pkg/observe/log"
	"github.com/hanzo-fi/go-libs/v5/pkg/query"
	"github.com/hanzo-fi/go-libs/v5/pkg/storage/bun/paginate"
	"github.com/hanzo-fi/go-libs/v5/pkg/types/time"
	"github.com/hanzo-fi/payments/pkg/domain/models"
	"github.com/hanzo-fi/payments/pkg/events"
	"github.com/gibson042/canonicaljson-go"
	"github.com/google/uuid"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
)

var (
	now              = time.Now()
	defaultConnector = models.Connector{
		ConnectorBase: models.ConnectorBase{
			ID: models.ConnectorID{
				Reference: uuid.New(),
				Provider:  "default",
			},
			Name:      "default",
			CreatedAt: now.Add(-60 * time.Minute).UTC().Time,
			Provider:  "default",
		},
		Config: []byte(`{}`),
	}

	defaultConnector2 = models.Connector{
		ConnectorBase: models.ConnectorBase{
			ID: models.ConnectorID{
				Reference: uuid.New(),
				Provider:  "default2",
			},
			Name:      "default2",
			CreatedAt: now.Add(-45 * time.Minute).UTC().Time,
			Provider:  "default2",
		},
		Config: []byte(`{}`),
	}

	defaultConnector3 = models.Connector{
		ConnectorBase: models.ConnectorBase{
			ID: models.ConnectorID{
				Reference: uuid.New(),
				Provider:  "default",
			},
			Name:      "default3",
			CreatedAt: now.Add(-30 * time.Minute).UTC().Time,
			Provider:  "default",
		},
		Config: []byte(`{}`),
	}
)

func upsertConnector(t *testing.T, ctx context.Context, storage Storage, connector models.Connector) {
	require.NoError(t, storage.ConnectorsInstall(ctx, connector, nil))
}

// withoutUpdatedAt zeroes the UpdatedAt field before structural comparison.
// The updated_at column is set by a DB trigger on every UPDATE (including the
// internal config-write during install), so its exact value is non-deterministic
// and should not be asserted in tests that focus on other fields.
func withoutUpdatedAt(c models.Connector) models.Connector {
	c.UpdatedAt = nil
	return c
}

func TestConnectorsInstall(t *testing.T) {
	t.Parallel()

	ctx := logging.TestingContext()
	store := newStore(t)
	defer store.Close()

	upsertConnector(t, ctx, store, defaultConnector)
	upsertConnector(t, ctx, store, defaultConnector2)

	t.Run("same id upsert", func(t *testing.T) {
		c := models.Connector{
			ConnectorBase: models.ConnectorBase{
				ID:        defaultConnector.ID,
				Name:      "test changed",
				CreatedAt: time.Now().UTC().Time,
				Provider:  "test",
			},
			Config: []byte(`{}`),
		}

		require.NoError(t, store.ConnectorsInstall(ctx, c, nil))

		connector, err := store.ConnectorsGet(ctx, c.ID)
		require.NoError(t, err)
		require.NotNil(t, connector)
		require.Equal(t, defaultConnector, withoutUpdatedAt(*connector))
	})

	t.Run("unique same upsert", func(t *testing.T) {
		c := models.Connector{
			ConnectorBase: models.ConnectorBase{
				ID: models.ConnectorID{
					Reference: uuid.New(),
					Provider:  "test",
				},
				Name:      "default",
				CreatedAt: now.Add(-23 * time.Minute).UTC().Time,
				Provider:  "test",
			},
			Config: []byte(`{}`),
		}

		require.Error(t, store.ConnectorsInstall(ctx, c, nil))
	})

	t.Run("outbox event created for connector reset", func(t *testing.T) {
		// Create an old connector that will be "reset"
		oldConnector := models.Connector{
			ConnectorBase: models.ConnectorBase{
				ID: models.ConnectorID{
					Reference: uuid.New(),
					Provider:  "reset-test",
				},
				Name:      "old connector",
				CreatedAt: now.Add(-60 * time.Minute).UTC().Time,
				Provider:  "reset-test",
			},
			Config: []byte(`{}`),
		}
		upsertConnector(t, ctx, store, oldConnector)

		// Create a new connector (simulating reset)
		newConnector := models.Connector{
			ConnectorBase: models.ConnectorBase{
				ID: models.ConnectorID{
					Reference: uuid.New(),
					Provider:  "reset-test",
				},
				Name:      "new connector",
				CreatedAt: now.Add(-1 * time.Minute).UTC().Time,
				Provider:  "reset-test",
			},
			Config: []byte(`{}`),
		}

		// Install with oldConnectorID (simulating reset)
		require.NoError(t, store.ConnectorsInstall(ctx, newConnector, &oldConnector.ID))

		// Verify outbox event was created
		pendingEvents, err := store.OutboxEventsPollPending(ctx, 100)
		require.NoError(t, err)

		// Find our event
		var ourEvent *models.OutboxEvent
		for i := range pendingEvents {
			if pendingEvents[i].EventType == events.EventTypeConnectorReset &&
				pendingEvents[i].EntityID == oldConnector.ID.String() {
				ourEvent = &pendingEvents[i]
				break
			}
		}
		require.NotNil(t, ourEvent, "expected outbox event for connector reset")

		// Verify event details
		assert.Equal(t, events.EventTypeConnectorReset, ourEvent.EventType)
		assert.Equal(t, models.OUTBOX_STATUS_PENDING, ourEvent.Status)
		assert.Equal(t, oldConnector.ID.String(), ourEvent.EntityID)
		assert.Equal(t, newConnector.ID, *ourEvent.ConnectorID)
		assert.Equal(t, 0, ourEvent.RetryCount)

		// Verify payload
		var payload map[string]interface{}
		err = json.Unmarshal(ourEvent.Payload, &payload)
		require.NoError(t, err)
		assert.Equal(t, oldConnector.ID.String(), payload["connectorID"])
		assert.NotNil(t, payload["createdAt"])

		// Verify idempotency key format
		expectedKeyPrefix := fmt.Sprintf("%s-", oldConnector.ID.String())
		assert.Contains(t, ourEvent.ID.EventIdempotencyKey, expectedKeyPrefix)
	})

	t.Run("no outbox event for normal install", func(t *testing.T) {
		// Count events before
		eventsBefore, err := store.OutboxEventsPollPending(ctx, 1000)
		require.NoError(t, err)
		countBefore := len(eventsBefore)

		// Create a new connector without oldConnectorID (normal install)
		normalConnector := models.Connector{
			ConnectorBase: models.ConnectorBase{
				ID: models.ConnectorID{
					Reference: uuid.New(),
					Provider:  "normal-test",
				},
				Name:      "normal connector",
				CreatedAt: now.Add(-1 * time.Minute).UTC().Time,
				Provider:  "normal-test",
			},
			Config: []byte(`{}`),
		}

		require.NoError(t, store.ConnectorsInstall(ctx, normalConnector, nil))

		// Verify no outbox event was created
		eventsAfter, err := store.OutboxEventsPollPending(ctx, 1000)
		require.NoError(t, err)
		countAfter := len(eventsAfter)

		// Should have same number of events (no new reset event)
		assert.Equal(t, countBefore, countAfter, "normal install should not create reset event")
	})
}

func TestConnectorsUninstall(t *testing.T) {
	t.Parallel()

	ctx := logging.TestingContext()
	store := newStore(t)
	defer store.Close()

	upsertConnector(t, ctx, store, defaultConnector)
	upsertConnector(t, ctx, store, defaultConnector2)

	t.Run("uninstall default connector", func(t *testing.T) {
		require.NoError(t, store.ConnectorsUninstall(ctx, defaultConnector.ID))

		connector, err := store.ConnectorsGet(ctx, defaultConnector.ID)
		require.Error(t, err)
		require.ErrorIs(t, err, ErrNotFound)
		require.Nil(t, connector)
	})

	t.Run("uninstall unknown connector", func(t *testing.T) {
		id := models.ConnectorID{
			Reference: uuid.New(),
			Provider:  "unknown",
		}

		require.NoError(t, store.ConnectorsUninstall(ctx, id))
	})
}

func TestConnectorsConfigUpdate(t *testing.T) {
	t.Parallel()

	ctx := logging.TestingContext()
	store := newStore(t)
	defer store.Close()

	upsertConnector(t, ctx, store, defaultConnector)
	upsertConnector(t, ctx, store, defaultConnector2)

	t.Run("same id upsert", func(t *testing.T) {
		config := json.RawMessage(`{"val":"new"}`)
		c := models.Connector{
			ConnectorBase: models.ConnectorBase{
				ID:   defaultConnector.ID,
				Name: "new name",
			},
			Config: config,
		}

		require.NoError(t, store.ConnectorsConfigUpdate(ctx, c))

		connector, err := store.ConnectorsGet(ctx, c.ID)
		require.NoError(t, err)
		require.NotNil(t, connector)
		assert.Equal(t, c.Name, connector.Name)
		assert.Equal(t, defaultConnector.CreatedAt, connector.CreatedAt)
		assert.Equal(t, defaultConnector.Provider, connector.Provider)
		assert.Equal(t, defaultConnector.ScheduledForDeletion, connector.ScheduledForDeletion)

		expectedData, err := canonicaljson.Marshal(config)
		require.NoError(t, err)
		data, err := canonicaljson.Marshal(connector.Config)
		require.NoError(t, err)
		assert.Equal(t, string(expectedData), string(data))
	})

	t.Run("connector doesn't exist yet", func(t *testing.T) {
		c := models.Connector{
			ConnectorBase: models.ConnectorBase{
				ID: models.ConnectorID{
					Reference: uuid.New(),
					Provider:  "test",
				},
			},
			Config: []byte(`{}`),
		}

		require.Error(t, store.ConnectorsConfigUpdate(ctx, c))
	})
}

func TestConnectorsScheduleForDeletion(t *testing.T) {
	t.Parallel()

	ctx := logging.TestingContext()
	store := newStore(t)
	defer store.Close()

	upsertConnector(t, ctx, store, defaultConnector)

	t.Run("schedule for deletion of unknown connector", func(t *testing.T) {
		id := models.ConnectorID{
			Reference: uuid.New(),
			Provider:  "unknown",
		}

		require.NoError(t, store.ConnectorsScheduleForDeletion(ctx, id))
	})

	t.Run("schedule for deletion of default connector", func(t *testing.T) {
		require.NoError(t, store.ConnectorsScheduleForDeletion(ctx, defaultConnector.ID))

		connector, err := store.ConnectorsGet(ctx, defaultConnector.ID)
		require.NoError(t, err)
		require.NotNil(t, connector)
		require.True(t, connector.ScheduledForDeletion)
	})
}

func TestConnectorsGet(t *testing.T) {
	t.Parallel()

	ctx := logging.TestingContext()
	store := newStore(t)
	defer store.Close()

	upsertConnector(t, ctx, store, defaultConnector)
	upsertConnector(t, ctx, store, defaultConnector2)

	t.Run("get connector", func(t *testing.T) {
		connector, err := store.ConnectorsGet(ctx, defaultConnector.ID)
		require.NoError(t, err)
		require.NotNil(t, connector)
		require.Equal(t, defaultConnector, withoutUpdatedAt(*connector))
	})

	t.Run("get unknown connector", func(t *testing.T) {
		id := models.ConnectorID{
			Reference: uuid.New(),
			Provider:  "unknown",
		}

		connector, err := store.ConnectorsGet(ctx, id)
		require.Error(t, err)
		require.ErrorIs(t, err, ErrNotFound)
		require.Nil(t, connector)
	})
}

func TestConnectorsList(t *testing.T) {
	t.Parallel()

	ctx := logging.TestingContext()
	store := newStore(t)
	defer store.Close()

	upsertConnector(t, ctx, store, defaultConnector)
	upsertConnector(t, ctx, store, defaultConnector2)
	upsertConnector(t, ctx, store, defaultConnector3)

	t.Run("list connectors by name", func(t *testing.T) {
		q := NewListConnectorsQuery(
			paginate.NewPaginatedQueryOptions(ConnectorQuery{}).
				WithPageSize(15).
				WithQueryBuilder(query.Match("name", "default")),
		)

		cursor, err := store.ConnectorsList(ctx, q)
		require.NoError(t, err)
		require.Len(t, cursor.Data, 1)
		require.False(t, cursor.HasMore)
		require.Empty(t, cursor.Next)
		require.Empty(t, cursor.Previous)
		require.Equal(t, defaultConnector, withoutUpdatedAt(cursor.Data[0]))
	})

	t.Run("list connectors by unknown name", func(t *testing.T) {
		q := NewListConnectorsQuery(
			paginate.NewPaginatedQueryOptions(ConnectorQuery{}).
				WithPageSize(15).
				WithQueryBuilder(query.Match("name", "unknown")),
		)

		cursor, err := store.ConnectorsList(ctx, q)
		require.NoError(t, err)
		require.Empty(t, cursor.Data)
		require.False(t, cursor.HasMore)
		require.Empty(t, cursor.Next)
		require.Empty(t, cursor.Previous)
	})

	t.Run("list connectors by provider", func(t *testing.T) {
		q := NewListConnectorsQuery(
			paginate.NewPaginatedQueryOptions(ConnectorQuery{}).
				WithPageSize(15).
				WithQueryBuilder(query.Match("provider", "default")),
		)

		cursor, err := store.ConnectorsList(ctx, q)
		require.NoError(t, err)
		require.Len(t, cursor.Data, 2)
		require.False(t, cursor.HasMore)
		require.Empty(t, cursor.Next)
		require.Empty(t, cursor.Previous)
		require.Equal(t, defaultConnector3, withoutUpdatedAt(cursor.Data[0]))
		require.Equal(t, defaultConnector, withoutUpdatedAt(cursor.Data[1]))
	})

	t.Run("list connectors by provider uppercase", func(t *testing.T) {
		q := NewListConnectorsQuery(
			paginate.NewPaginatedQueryOptions(ConnectorQuery{}).
				WithPageSize(15).
				WithQueryBuilder(query.Match("provider", "DEFAULT")),
		)

		cursor, err := store.ConnectorsList(ctx, q)
		require.NoError(t, err)
		require.Len(t, cursor.Data, 2)
		require.False(t, cursor.HasMore)
		require.Empty(t, cursor.Next)
		require.Empty(t, cursor.Previous)
		require.Equal(t, defaultConnector3, withoutUpdatedAt(cursor.Data[0]))
		require.Equal(t, defaultConnector, withoutUpdatedAt(cursor.Data[1]))
	})

	t.Run("list connectors by provider with wrong type", func(t *testing.T) {
		q := NewListConnectorsQuery(
			paginate.NewPaginatedQueryOptions(ConnectorQuery{}).
				WithPageSize(15).
				WithQueryBuilder(query.Match("provider", 1)),
		)

		cursor, err := store.ConnectorsList(ctx, q)
		require.Error(t, err)
		require.Nil(t, cursor)
	})

	t.Run("list connectors by unknown provider", func(t *testing.T) {
		q := NewListConnectorsQuery(
			paginate.NewPaginatedQueryOptions(ConnectorQuery{}).
				WithPageSize(15).
				WithQueryBuilder(query.Match("provider", "unknown")),
		)

		cursor, err := store.ConnectorsList(ctx, q)
		require.NoError(t, err)
		require.Empty(t, cursor.Data)
		require.False(t, cursor.HasMore)
		require.Empty(t, cursor.Next)
		require.Empty(t, cursor.Previous)
	})

	t.Run("list connectors by id", func(t *testing.T) {
		q := NewListConnectorsQuery(
			paginate.NewPaginatedQueryOptions(ConnectorQuery{}).
				WithPageSize(15).
				WithQueryBuilder(query.Match("id", defaultConnector3.ID.String())),
		)

		cursor, err := store.ConnectorsList(ctx, q)
		require.NoError(t, err)
		require.Len(t, cursor.Data, 1)
		require.False(t, cursor.HasMore)
		require.Empty(t, cursor.Next)
		require.Empty(t, cursor.Previous)
		require.Equal(t, defaultConnector3, withoutUpdatedAt(cursor.Data[0]))
	})

	t.Run("list connectors by unknown id", func(t *testing.T) {
		q := NewListConnectorsQuery(
			paginate.NewPaginatedQueryOptions(ConnectorQuery{}).
				WithPageSize(15).
				WithQueryBuilder(query.Match("id", "unknown")),
		)

		cursor, err := store.ConnectorsList(ctx, q)
		require.NoError(t, err)
		require.Empty(t, cursor.Data)
		require.False(t, cursor.HasMore)
		require.Empty(t, cursor.Next)
		require.Empty(t, cursor.Previous)
	})

	t.Run("list connectors test cursor", func(t *testing.T) {
		q := NewListConnectorsQuery(
			paginate.NewPaginatedQueryOptions(ConnectorQuery{}).
				WithPageSize(1),
		)

		cursor, err := store.ConnectorsList(ctx, q)
		require.NoError(t, err)
		require.Len(t, cursor.Data, 1)
		require.True(t, cursor.HasMore)
		require.NotEmpty(t, cursor.Next)
		require.Empty(t, cursor.Previous)
		require.Equal(t, defaultConnector3, withoutUpdatedAt(cursor.Data[0]))

		err = paginate.UnmarshalCursor(cursor.Next, &q)
		require.NoError(t, err)
		cursor, err = store.ConnectorsList(ctx, q)
		require.NoError(t, err)
		require.Len(t, cursor.Data, 1)
		require.True(t, cursor.HasMore)
		require.NotEmpty(t, cursor.Next)
		require.NotEmpty(t, cursor.Previous)
		require.Equal(t, defaultConnector2, withoutUpdatedAt(cursor.Data[0]))

		err = paginate.UnmarshalCursor(cursor.Next, &q)
		require.NoError(t, err)
		cursor, err = store.ConnectorsList(ctx, q)
		require.NoError(t, err)
		require.Len(t, cursor.Data, 1)
		require.False(t, cursor.HasMore)
		require.Empty(t, cursor.Next)
		require.NotEmpty(t, cursor.Previous)
		require.Equal(t, defaultConnector, withoutUpdatedAt(cursor.Data[0]))

		err = paginate.UnmarshalCursor(cursor.Previous, &q)
		require.NoError(t, err)
		cursor, err = store.ConnectorsList(ctx, q)
		require.NoError(t, err)
		require.Len(t, cursor.Data, 1)
		require.True(t, cursor.HasMore)
		require.NotEmpty(t, cursor.Next)
		require.NotEmpty(t, cursor.Previous)
		require.Equal(t, defaultConnector2, withoutUpdatedAt(cursor.Data[0]))

		err = paginate.UnmarshalCursor(cursor.Previous, &q)
		require.NoError(t, err)
		cursor, err = store.ConnectorsList(ctx, q)
		require.NoError(t, err)
		require.Len(t, cursor.Data, 1)
		require.True(t, cursor.HasMore)
		require.NotEmpty(t, cursor.Next)
		require.Empty(t, cursor.Previous)
		require.Equal(t, defaultConnector3, withoutUpdatedAt(cursor.Data[0]))
	})
}