payments/internal/storage/accounts_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/pointer"
	"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/google/uuid"
	"github.com/pkg/errors"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
)

func defaultAccounts() []models.Account {
	return []models.Account{
		{
			ID: models.AccountID{
				Reference:   "test1",
				ConnectorID: defaultConnector.ID,
			},
			ConnectorID:  defaultConnector.ID,
			Connector:    defaultConnector.Base(),
			Reference:    "test1",
			CreatedAt:    now.Add(-60 * time.Minute).UTC().Time,
			Type:         models.ACCOUNT_TYPE_INTERNAL,
			Name:         pointer.For("test1"),
			DefaultAsset: pointer.For("USD/2"),
			Metadata: map[string]string{
				"foo": "bar",
			},
			Raw: []byte(`{}`),
		},
		{
			ID: models.AccountID{
				Reference:   "test2",
				ConnectorID: defaultConnector.ID,
			},
			ConnectorID: defaultConnector.ID,
			Connector:   defaultConnector.Base(),
			Reference:   "test2",
			CreatedAt:   now.Add(-30 * time.Minute).UTC().Time,
			Type:        models.ACCOUNT_TYPE_INTERNAL,
			Metadata: map[string]string{
				"foo2": "bar2",
			},
			Raw: []byte(`{}`),
		},
		{
			ID: models.AccountID{
				Reference:   "test3",
				ConnectorID: defaultConnector.ID,
			},
			ConnectorID: defaultConnector.ID,
			Connector:   defaultConnector.Base(),
			Reference:   "test3",
			CreatedAt:   now.Add(-45 * time.Minute).UTC().Time,
			Type:        models.ACCOUNT_TYPE_EXTERNAL,
			Name:        pointer.For("test3"),
			Metadata: map[string]string{
				"foo3": "bar3",
			},
			Raw: []byte(`{}`),
		},
	}
}

func defaultAccounts2() []models.Account {
	return []models.Account{
		{
			ID: models.AccountID{
				Reference:   "test1",
				ConnectorID: defaultConnector2.ID,
			},
			ConnectorID:  defaultConnector2.ID,
			Connector:    defaultConnector2.Base(),
			Reference:    "test1",
			CreatedAt:    now.Add(-55 * time.Minute).UTC().Time,
			Type:         models.ACCOUNT_TYPE_INTERNAL,
			Name:         pointer.For("test1"),
			DefaultAsset: pointer.For("USD/2"),
			Metadata: map[string]string{
				"foo5": "bar5",
			},
			Raw: []byte(`{}`),
		},
	}
}

func defaultAccounts3() []models.Account {
	createdAt := now.UTC().Truncate(time.Minute).Add(-2 * time.Minute).UTC()
	return []models.Account{
		{
			ID: models.AccountID{
				Reference:   "sort-test",
				ConnectorID: defaultConnector2.ID,
			},
			ConnectorID:  defaultConnector2.ID,
			Connector:    defaultConnector2.Base(),
			Reference:    "sort-test",
			CreatedAt:    createdAt,
			Type:         models.ACCOUNT_TYPE_INTERNAL,
			Name:         pointer.For("sort-test"),
			DefaultAsset: pointer.For("EUR/2"),
			Metadata: map[string]string{
				"unrelated": "keyval",
			},
			Raw: []byte(`{}`),
		},
		{
			ID: models.AccountID{
				Reference:   "sort-test2",
				ConnectorID: defaultConnector2.ID,
			},
			ConnectorID:  defaultConnector2.ID,
			Connector:    defaultConnector2.Base(),
			Reference:    "sort-test2",
			CreatedAt:    createdAt,
			Type:         models.ACCOUNT_TYPE_INTERNAL,
			Name:         pointer.For("sort-test2"),
			DefaultAsset: pointer.For("EUR/2"),
			Metadata: map[string]string{
				"metadata": "keyval",
			},
			Raw: []byte(`{}`),
		},
	}
}

func upsertAccounts(t *testing.T, ctx context.Context, storage Storage, accounts []models.Account) {
	require.NoError(t, storage.AccountsUpsert(ctx, accounts))
}

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

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

	upsertConnector(t, ctx, store, defaultConnector)
	upsertAccounts(t, ctx, store, defaultAccounts())
	cleanupOutboxHelper(ctx, store)() // Remove the outbox events created by updserting accounts

	t.Run("upsert empty list", func(t *testing.T) {
		require.NoError(t, store.AccountsUpsert(ctx, []models.Account{}))
	})

	t.Run("same id insert", func(t *testing.T) {
		defer cleanupOutboxHelper(ctx, store)() // Also clean up after test

		id := models.AccountID{
			Reference:   "test1",
			ConnectorID: defaultConnector.ID,
		}

		// Same account I but different fields
		acc := models.Account{
			ID:           id,
			ConnectorID:  defaultConnector.ID,
			Connector:    defaultConnector.Base(),
			Reference:    "test1",
			CreatedAt:    now.Add(-12 * time.Minute).UTC().Time,
			Type:         models.ACCOUNT_TYPE_EXTERNAL,
			Name:         pointer.For("changed"),
			DefaultAsset: pointer.For("EUR"),
			Metadata: map[string]string{
				"foo4": "bar4",
			},
			Raw: []byte(`{}`),
		}

		require.NoError(t, store.AccountsUpsert(ctx, []models.Account{acc}))

		// Check that account was not updated
		account, err := store.AccountsGet(ctx, id)
		require.NoError(t, err)

		// Accounts should not have changed
		require.Equal(t, defaultAccounts()[0], *account)

		// Verify outbox events were not created (account already existed)
		pendingEvents, err := store.OutboxEventsPollPending(ctx, 10)
		require.NoError(t, err)
		// Should have no new events since the account already existed
		assert.Len(t, pendingEvents, 0, "No outbox events should be created for existing accounts")
	})

	t.Run("unknown connector id", func(t *testing.T) {
		defer cleanupOutboxHelper(ctx, store)()

		unknownConnectorID := models.ConnectorID{
			Reference: uuid.New(),
			Provider:  "unknown",
		}

		acc := models.Account{
			ID: models.AccountID{
				Reference:   "test_unknown",
				ConnectorID: unknownConnectorID,
			},
			ConnectorID:  unknownConnectorID,
			Reference:    "test_unknown",
			CreatedAt:    now.Add(-12 * time.Minute).UTC().Time,
			Type:         models.ACCOUNT_TYPE_EXTERNAL,
			Name:         pointer.For("changed"),
			DefaultAsset: pointer.For("EUR"),
			Metadata: map[string]string{
				"foo4": "bar4",
			},
			Raw: []byte(`{}`),
		}

		require.Error(t, store.AccountsUpsert(ctx, []models.Account{acc}))

		// Verify outbox events were not created
		pendingEvents, err := store.OutboxEventsPollPending(ctx, 10)
		require.NoError(t, err)
		require.Len(t, pendingEvents, 0)
	})

	t.Run("outbox events created for new accounts", func(t *testing.T) {
		defer cleanupOutboxHelper(ctx, store)()

		// Create new accounts
		newAccounts := []models.Account{
			{
				ID: models.AccountID{
					Reference:   "outbox-test-1",
					ConnectorID: defaultConnector.ID,
				},
				ConnectorID:  defaultConnector.ID,
				Connector:    defaultConnector.Base(),
				Reference:    "outbox-test-1",
				CreatedAt:    time.Now().UTC().Time,
				Type:         models.ACCOUNT_TYPE_INTERNAL,
				Name:         pointer.For("Outbox Test 1"),
				DefaultAsset: pointer.For("USD/2"),
				Metadata: map[string]string{
					"test": "outbox",
				},
				Raw: []byte(`{}`),
			},
			{
				ID: models.AccountID{
					Reference:   "outbox-test-2",
					ConnectorID: defaultConnector.ID,
				},
				ConnectorID:  defaultConnector.ID,
				Connector:    defaultConnector.Base(),
				Reference:    "outbox-test-2",
				CreatedAt:    time.Now().UTC().Time,
				Type:         models.ACCOUNT_TYPE_EXTERNAL,
				Name:         pointer.For("Outbox Test 2"),
				DefaultAsset: pointer.For("EUR/2"),
				Metadata: map[string]string{
					"test": "outbox",
				},
				Raw: []byte(`{}`),
			},
		}

		// Create a set of expected idempotency keys
		expectedKeys := make(map[string]bool)
		for _, account := range newAccounts {
			expectedKeys[account.IdempotencyKey()] = true
		}

		// Insert accounts
		require.NoError(t, store.AccountsUpsert(ctx, newAccounts))

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

		// Filter events to only those we just created
		ourEvents := make([]models.OutboxEvent, 0)
		for _, event := range pendingEvents {
			if event.EventType == events.EventTypeSavedAccounts && expectedKeys[event.ID.EventIdempotencyKey] {
				ourEvents = append(ourEvents, event)
			}
		}
		require.Len(t, ourEvents, 2, "expected 2 outbox events for 2 new accounts")

		// Create a map of expected accounts by idempotency key for easier lookup
		expectedAccountsByKey := make(map[string]models.Account)
		for _, account := range newAccounts {
			expectedAccountsByKey[account.IdempotencyKey()] = account
		}

		// Check event details
		for _, event := range ourEvents {
			assert.Equal(t, events.EventTypeSavedAccounts, event.EventType)
			assert.Equal(t, models.OUTBOX_STATUS_PENDING, event.Status)
			assert.Equal(t, defaultConnector.ID, *event.ConnectorID)
			assert.Equal(t, 0, event.RetryCount)
			assert.Nil(t, event.Error)
			assert.NotEmpty(t, event.ID.EventIdempotencyKey)

			// Find the matching account by idempotency key
			expectedAccount, found := expectedAccountsByKey[event.ID.EventIdempotencyKey]
			require.True(t, found, "event idempotency key should match one of the accounts")

			// Verify payload contains account data
			var payload map[string]interface{}
			err := json.Unmarshal(event.Payload, &payload)
			require.NoError(t, err)
			assert.Equal(t, expectedAccount.ID.String(), payload["id"])
			assert.Contains(t, payload, "name")
			assert.Contains(t, payload, "type")
			assert.Equal(t, expectedAccount.ConnectorID.String(), payload["connectorID"])

			// Verify EntityID matches account ID
			assert.Equal(t, expectedAccount.ID.String(), event.EntityID)

			// Verify idempotency key matches account
			assert.Equal(t, expectedAccount.IdempotencyKey(), event.ID.EventIdempotencyKey)
		}
	})

	t.Run("no outbox events for existing accounts", func(t *testing.T) {
		defer cleanupOutboxHelper(ctx, store)()

		// Try to insert existing accounts (should not create outbox events due to ON CONFLICT DO NOTHING)
		require.NoError(t, store.AccountsUpsert(ctx, defaultAccounts()))

		// Verify no outbox events were created
		pendingEvents, err := store.OutboxEventsPollPending(ctx, 10)
		require.NoError(t, err)
		assert.Len(t, pendingEvents, 0)
	})

	t.Run("rollback on foreign key violation", func(t *testing.T) {
		defer cleanupOutboxHelper(ctx, store)()

		upsertConnector(t, ctx, store, defaultConnector)

		// Count existing accounts
		accountsBefore, err := store.AccountsList(ctx, NewListAccountsQuery(paginate.NewPaginatedQueryOptions(AccountQuery{}).WithPageSize(1000)))
		require.NoError(t, err)
		countBefore := len(accountsBefore.Data)

		// Create an account with an invalid connector ID that doesn't exist
		invalidConnectorID := models.ConnectorID{Reference: uuid.New(), Provider: "non-existent-provider"}
		invalidAccount := models.Account{
			ID:          models.AccountID{Reference: "invalid-ref", ConnectorID: invalidConnectorID},
			ConnectorID: invalidConnectorID,
			CreatedAt:   time.Now().UTC().Time,
			Reference:   "invalid-ref",
			Type:        models.ACCOUNT_TYPE_EXTERNAL,
		}

		// Attempt to upsert - should fail due to foreign key violation
		err = store.AccountsUpsert(ctx, []models.Account{invalidAccount})
		require.Error(t, err)

		// Verify no account was inserted
		accountsAfter, err := store.AccountsList(ctx, NewListAccountsQuery(paginate.NewPaginatedQueryOptions(AccountQuery{}).WithPageSize(1000)))
		require.NoError(t, err)
		assert.Equal(t, countBefore, len(accountsAfter.Data), "no accounts should be inserted on error")

		// Verify no outbox events were created
		pendingEvents, err := store.OutboxEventsPollPending(ctx, 100)
		require.NoError(t, err)
		for _, event := range pendingEvents {
			assert.NotEqual(t, invalidAccount.ID.String(), event.EntityID, "no outbox event should be created for failed insert")
		}
	})
}

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

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

	upsertConnector(t, ctx, store, defaultConnector)
	upsertAccounts(t, ctx, store, defaultAccounts())

	t.Run("get account", func(t *testing.T) {
		for _, acc := range defaultAccounts() {
			account, err := store.AccountsGet(ctx, acc.ID)
			require.NoError(t, err)
			require.Equal(t, acc, *account)
		}
	})

	t.Run("get unknown account", func(t *testing.T) {
		acc := models.AccountID{
			Reference:   "unknown",
			ConnectorID: defaultConnector.ID,
		}

		account, err := store.AccountsGet(ctx, acc)
		require.Error(t, err)
		require.ErrorIs(t, err, ErrNotFound)
		require.Nil(t, account)
	})
}

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

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

	upsertConnector(t, ctx, store, defaultConnector)
	upsertConnector(t, ctx, store, defaultConnector2)
	upsertAccounts(t, ctx, store, defaultAccounts())
	upsertAccounts(t, ctx, store, defaultAccounts2())

	t.Run("delete account from unknown connector", func(t *testing.T) {
		unknownConnectorID := models.ConnectorID{
			Reference: uuid.New(),
			Provider:  "unknown",
		}

		require.NoError(t, store.AccountsDeleteFromConnectorID(ctx, unknownConnectorID))

		for _, acc := range defaultAccounts() {
			account, err := store.AccountsGet(ctx, acc.ID)
			require.NoError(t, err)
			require.Equal(t, acc, *account)
		}

		for _, acc := range defaultAccounts2() {
			account, err := store.AccountsGet(ctx, acc.ID)
			require.NoError(t, err)
			require.Equal(t, acc, *account)
		}
	})

	t.Run("delete account from default connector", func(t *testing.T) {
		require.NoError(t, store.AccountsDeleteFromConnectorID(ctx, defaultConnector.ID))

		for _, acc := range defaultAccounts() {
			account, err := store.AccountsGet(ctx, acc.ID)
			require.Error(t, err)
			require.Nil(t, account)
			require.ErrorIs(t, err, ErrNotFound)
		}

		for _, acc := range defaultAccounts2() {
			account, err := store.AccountsGet(ctx, acc.ID)
			require.NoError(t, err)
			require.Equal(t, acc, *account)
		}
	})

}

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

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

	upsertConnector(t, ctx, store, defaultConnector)
	upsertConnector(t, ctx, store, defaultConnector2)
	upsertAccounts(t, ctx, store, defaultAccounts())
	upsertAccounts(t, ctx, store, defaultAccounts2())
	upsertAccounts(t, ctx, store, defaultAccounts3())

	t.Run("list accounts by reference", func(t *testing.T) {
		q := NewListAccountsQuery(
			paginate.NewPaginatedQueryOptions(AccountQuery{}).
				WithPageSize(15).
				WithQueryBuilder(query.Match("reference", "test1")),
		)

		cursor, err := store.AccountsList(ctx, q)
		require.NoError(t, err)
		require.Len(t, cursor.Data, 2)
		require.False(t, cursor.HasMore)
		require.Equal(t, defaultAccounts2()[0], cursor.Data[0])
		require.Equal(t, defaultAccounts()[0], cursor.Data[1])
	})

	t.Run("list accounts by reference 2", func(t *testing.T) {
		q := NewListAccountsQuery(
			paginate.NewPaginatedQueryOptions(AccountQuery{}).
				WithPageSize(15).
				WithQueryBuilder(query.Match("reference", "test2")),
		)
		cursor, err := store.AccountsList(ctx, q)
		require.NoError(t, err)
		require.Len(t, cursor.Data, 1)
		require.False(t, cursor.HasMore)
		require.Equal(t, defaultAccounts()[1], cursor.Data[0])
	})

	t.Run("list accounts by unknown reference", func(t *testing.T) {
		q := NewListAccountsQuery(
			paginate.NewPaginatedQueryOptions(AccountQuery{}).
				WithPageSize(15).
				WithQueryBuilder(query.Match("reference", "unknown")),
		)

		cursor, err := store.AccountsList(ctx, q)
		require.NoError(t, err)
		require.Len(t, cursor.Data, 0)
		require.False(t, cursor.HasMore)
	})

	t.Run("list accounts by connector id", func(t *testing.T) {
		q := NewListAccountsQuery(
			paginate.NewPaginatedQueryOptions(AccountQuery{}).
				WithPageSize(15).
				WithQueryBuilder(query.Match("connector_id", defaultConnector.ID)),
		)
		accounts := defaultAccounts()
		cursor, err := store.AccountsList(ctx, q)
		require.NoError(t, err)
		require.Len(t, cursor.Data, 3)
		require.False(t, cursor.HasMore)
		require.Equal(t, accounts[1], cursor.Data[0])
		require.Equal(t, accounts[2], cursor.Data[1])
		require.Equal(t, accounts[0], cursor.Data[2])
	})

	t.Run("list accounts by connector id 2", func(t *testing.T) {
		q := NewListAccountsQuery(
			paginate.NewPaginatedQueryOptions(AccountQuery{}).
				WithPageSize(15).
				WithQueryBuilder(query.Match("connector_id", defaultConnector2.ID)),
		)
		cursor, err := store.AccountsList(ctx, q)
		require.NoError(t, err)
		require.Len(t, cursor.Data, 3)
		require.False(t, cursor.HasMore)
		require.Equal(t, defaultAccounts3()[1], cursor.Data[0])
		require.Equal(t, defaultAccounts3()[0], cursor.Data[1])
		require.Equal(t, defaultAccounts2()[0], cursor.Data[2])
	})

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

		q := NewListAccountsQuery(
			paginate.NewPaginatedQueryOptions(AccountQuery{}).
				WithPageSize(15).
				WithQueryBuilder(query.Match("connector_id", unknownConnectorID)),
		)
		cursor, err := store.AccountsList(ctx, q)
		require.NoError(t, err)
		require.Len(t, cursor.Data, 0)
		require.False(t, cursor.HasMore)
	})

	t.Run("list accounts by id", func(t *testing.T) {
		q := NewListAccountsQuery(
			paginate.NewPaginatedQueryOptions(AccountQuery{}).
				WithPageSize(15).
				WithQueryBuilder(query.Match("id", defaultAccounts2()[0].ID.String())),
		)

		cursor, err := store.AccountsList(ctx, q)
		require.NoError(t, err)
		require.Len(t, cursor.Data, 1)
		require.False(t, cursor.HasMore)
		require.Equal(t, defaultAccounts2()[0], cursor.Data[0])
	})

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

		cursor, err := store.AccountsList(ctx, q)
		require.NoError(t, err)
		require.Len(t, cursor.Data, 0)
		require.False(t, cursor.HasMore)
	})

	t.Run("list accounts by type", func(t *testing.T) {
		q := NewListAccountsQuery(
			paginate.NewPaginatedQueryOptions(AccountQuery{}).
				WithPageSize(15).
				WithQueryBuilder(query.Match("type", models.ACCOUNT_TYPE_INTERNAL)),
		)
		accounts := defaultAccounts()
		accounts2 := defaultAccounts2()
		accounts3 := defaultAccounts3()

		cursor, err := store.AccountsList(ctx, q)
		require.NoError(t, err)
		require.Len(t, cursor.Data, 5)
		require.False(t, cursor.HasMore)
		require.Equal(t, accounts3[1], cursor.Data[0])
		require.Equal(t, accounts3[0], cursor.Data[1])
		require.Equal(t, accounts[1], cursor.Data[2])
		require.Equal(t, accounts2[0], cursor.Data[3])
		require.Equal(t, accounts[0], cursor.Data[4])
	})

	t.Run("list accounts by unknown type", func(t *testing.T) {
		q := NewListAccountsQuery(
			paginate.NewPaginatedQueryOptions(AccountQuery{}).
				WithPageSize(15).
				WithQueryBuilder(query.Match("type", "unknown")),
		)

		cursor, err := store.AccountsList(ctx, q)
		require.NoError(t, err)
		require.Len(t, cursor.Data, 0)
		require.False(t, cursor.HasMore)
	})

	t.Run("list accounts by default asset", func(t *testing.T) {
		q := NewListAccountsQuery(
			paginate.NewPaginatedQueryOptions(AccountQuery{}).
				WithPageSize(15).
				WithQueryBuilder(query.Match("default_asset", "USD/2")),
		)
		accounts := defaultAccounts()
		accounts2 := defaultAccounts2()

		cursor, err := store.AccountsList(ctx, q)
		require.NoError(t, err)
		require.Len(t, cursor.Data, 2)
		require.False(t, cursor.HasMore)
		require.Equal(t, accounts2[0], cursor.Data[0])
		require.Equal(t, accounts[0], cursor.Data[1])
	})

	t.Run("list accounts by unknown default asset", func(t *testing.T) {
		q := NewListAccountsQuery(
			paginate.NewPaginatedQueryOptions(AccountQuery{}).
				WithPageSize(15).
				WithQueryBuilder(query.Match("default_asset", "unknown")),
		)

		cursor, err := store.AccountsList(ctx, q)
		require.NoError(t, err)
		require.Len(t, cursor.Data, 0)
		require.False(t, cursor.HasMore)
	})

	t.Run("list accounts by name", func(t *testing.T) {
		q := NewListAccountsQuery(
			paginate.NewPaginatedQueryOptions(AccountQuery{}).
				WithPageSize(15).
				WithQueryBuilder(query.Match("name", "test1")),
		)

		accounts := defaultAccounts()
		accounts2 := defaultAccounts2()
		cursor, err := store.AccountsList(ctx, q)
		require.NoError(t, err)
		require.Len(t, cursor.Data, 2)
		require.False(t, cursor.HasMore)
		require.Equal(t, accounts2[0], cursor.Data[0])
		require.Equal(t, accounts[0], cursor.Data[1])
	})

	t.Run("list accounts by name 2", func(t *testing.T) {
		q := NewListAccountsQuery(
			paginate.NewPaginatedQueryOptions(AccountQuery{}).
				WithPageSize(15).
				WithQueryBuilder(query.Match("name", "test3")),
		)

		cursor, err := store.AccountsList(ctx, q)
		require.NoError(t, err)
		require.Len(t, cursor.Data, 1)
		require.False(t, cursor.HasMore)
		require.Equal(t, defaultAccounts()[2], cursor.Data[0])
	})

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

		cursor, err := store.AccountsList(ctx, q)
		require.NoError(t, err)
		require.Len(t, cursor.Data, 0)
		require.False(t, cursor.HasMore)
	})

	t.Run("list accounts by metadata", func(t *testing.T) {
		q := NewListAccountsQuery(
			paginate.NewPaginatedQueryOptions(AccountQuery{}).
				WithPageSize(15).
				WithQueryBuilder(query.Match("metadata[foo]", "bar")),
		)

		cursor, err := store.AccountsList(ctx, q)
		require.NoError(t, err)
		require.Len(t, cursor.Data, 1)
		require.False(t, cursor.HasMore)
		require.Equal(t, defaultAccounts()[0], cursor.Data[0])
	})

	t.Run("list accounts by unknown metadata", func(t *testing.T) {
		q := NewListAccountsQuery(
			paginate.NewPaginatedQueryOptions(AccountQuery{}).
				WithPageSize(15).
				WithQueryBuilder(query.Match("metadata[foo]", "unknown")),
		)

		cursor, err := store.AccountsList(ctx, q)
		require.NoError(t, err)
		require.Len(t, cursor.Data, 0)
		require.False(t, cursor.HasMore)
	})

	t.Run("wrong query builder operator with metadata", func(t *testing.T) {
		q := NewListAccountsQuery(
			paginate.NewPaginatedQueryOptions(AccountQuery{}).
				WithPageSize(15).
				WithQueryBuilder(query.Lt("metadata[foo]", "unknown")),
		)

		cursor, err := store.AccountsList(ctx, q)
		require.Error(t, err)
		require.Nil(t, cursor)
		assert.True(t, errors.Is(err, ErrValidation))
		assert.Regexp(t, "metadata\\[foo\\]", err.Error())
	})

	t.Run("query builder unknown key", func(t *testing.T) {
		q := NewListAccountsQuery(
			paginate.NewPaginatedQueryOptions(AccountQuery{}).
				WithPageSize(15).
				WithQueryBuilder(query.Match("unknown", "unknown")),
		)

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

	t.Run("list accounts test cursor", func(t *testing.T) {
		q := NewListAccountsQuery(
			paginate.NewPaginatedQueryOptions(AccountQuery{}).
				WithPageSize(1),
		)
		accounts := defaultAccounts()
		accounts2 := defaultAccounts2()
		accounts3 := defaultAccounts3()

		cursor, err := store.AccountsList(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, accounts3[1], cursor.Data[0])

		err = paginate.UnmarshalCursor(cursor.Next, &q)
		require.NoError(t, err)
		cursor, err = store.AccountsList(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, accounts3[0], cursor.Data[0])

		err = paginate.UnmarshalCursor(cursor.Next, &q)
		require.NoError(t, err)
		cursor, err = store.AccountsList(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, accounts[1], cursor.Data[0])

		err = paginate.UnmarshalCursor(cursor.Next, &q)
		require.NoError(t, err)
		cursor, err = store.AccountsList(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, accounts[2], cursor.Data[0])

		err = paginate.UnmarshalCursor(cursor.Next, &q)
		require.NoError(t, err)
		cursor, err = store.AccountsList(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, accounts2[0], cursor.Data[0])

		err = paginate.UnmarshalCursor(cursor.Next, &q)
		require.NoError(t, err)
		cursor, err = store.AccountsList(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, accounts[0], cursor.Data[0])

		err = paginate.UnmarshalCursor(cursor.Previous, &q)
		require.NoError(t, err)
		cursor, err = store.AccountsList(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, accounts2[0], cursor.Data[0])

		err = paginate.UnmarshalCursor(cursor.Previous, &q)
		require.NoError(t, err)
		cursor, err = store.AccountsList(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, accounts[2], cursor.Data[0])
	})
}
func TestAccountsDeleteFromConnectorIDBatch(t *testing.T) {
	t.Parallel()

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

	upsertConnector(t, ctx, store, defaultConnector)

	t.Run("invalid batchSize zero", func(t *testing.T) {
		_, err := store.AccountsDeleteFromConnectorIDBatch(ctx, defaultConnector.ID, 0)
		require.Error(t, err)
		require.ErrorIs(t, err, ErrValidation)
		require.Contains(t, err.Error(), "invalid batchSize 0")
		require.Contains(t, err.Error(), defaultConnector.ID.String())
	})

	t.Run("invalid batchSize negative", func(t *testing.T) {
		_, err := store.AccountsDeleteFromConnectorIDBatch(ctx, defaultConnector.ID, -1)
		require.Error(t, err)
		require.ErrorIs(t, err, ErrValidation)
		require.Contains(t, err.Error(), "invalid batchSize -1")
		require.Contains(t, err.Error(), defaultConnector.ID.String())
	})

	t.Run("delete batch from unknown connector", func(t *testing.T) {
		rowsAffected, err := store.AccountsDeleteFromConnectorIDBatch(ctx, models.ConnectorID{
			Reference: uuid.New(),
			Provider:  "unknown",
		}, 10)
		require.NoError(t, err)
		require.Equal(t, 0, rowsAffected)
	})

	t.Run("delete batch with no accounts", func(t *testing.T) {
		rowsAffected, err := store.AccountsDeleteFromConnectorIDBatch(ctx, defaultConnector.ID, 10)
		require.NoError(t, err)
		require.Equal(t, 0, rowsAffected)
	})

	t.Run("delete single batch smaller than batch size", func(t *testing.T) {
		upsertAccounts(t, ctx, store, defaultAccounts())

		rowsAffected, err := store.AccountsDeleteFromConnectorIDBatch(ctx, defaultConnector.ID, 10)
		require.NoError(t, err)
		require.Equal(t, 3, rowsAffected)

		// Verify all accounts were deleted
		for _, a := range defaultAccounts() {
			_, err := store.AccountsGet(ctx, a.ID)
			require.Error(t, err)
			require.ErrorIs(t, err, ErrNotFound)
		}
	})

	t.Run("delete multiple batches", func(t *testing.T) {
		// Create a connector with more accounts
		connector2 := models.Connector{
			ConnectorBase: models.ConnectorBase{
				ID: models.ConnectorID{
					Reference: uuid.New(),
					Provider:  "test2",
				},
				Name:      "test2",
				Provider:  "test2",
				CreatedAt: now.Add(-50 * time.Minute).UTC().Time,
			},
			ScheduledForDeletion: false,
			Config:               []byte(`{}`),
		}
		upsertConnector(t, ctx, store, connector2)

		// Create 5 accounts for this connector
		accounts := make([]models.Account, 5)
		for i := 0; i < 5; i++ {
			accounts[i] = models.Account{
				ID: models.AccountID{
					Reference:   fmt.Sprintf("batch-test-%d", i),
					ConnectorID: connector2.ID,
				},
				ConnectorID: connector2.ID,
				Reference:   fmt.Sprintf("batch-test-%d", i),
				CreatedAt:   now.Add(-40 * time.Minute).UTC().Time,
				Type:        models.ACCOUNT_TYPE_INTERNAL,
				Raw:         []byte(`{}`),
			}
		}
		upsertAccounts(t, ctx, store, accounts)

		// Delete in batches of 2
		totalDeleted := 0
		for {
			rowsAffected, err := store.AccountsDeleteFromConnectorIDBatch(ctx, connector2.ID, 2)
			require.NoError(t, err)
			if rowsAffected == 0 {
				break
			}
			totalDeleted += rowsAffected
		}

		require.Equal(t, 5, totalDeleted)

		// Verify all accounts were deleted
		for _, a := range accounts {
			_, err := store.AccountsGet(ctx, a.ID)
			require.Error(t, err)
			require.ErrorIs(t, err, ErrNotFound)
		}
	})

	t.Run("delete batch only affects specified connector", func(t *testing.T) {
		// Create two connectors with accounts
		connector3 := models.Connector{
			ConnectorBase: models.ConnectorBase{
				ID: models.ConnectorID{
					Reference: uuid.New(),
					Provider:  "test3",
				},
				Name:      "test3",
				Provider:  "test3",
				CreatedAt: now.Add(-45 * time.Minute).UTC().Time,
			},
			ScheduledForDeletion: false,
			Config:               []byte(`{}`),
		}
		upsertConnector(t, ctx, store, connector3)

		connector4 := models.Connector{
			ConnectorBase: models.ConnectorBase{
				ID: models.ConnectorID{
					Reference: uuid.New(),
					Provider:  "test4",
				},
				Name:      "test4",
				Provider:  "test4",
				CreatedAt: now.Add(-45 * time.Minute).UTC().Time,
			},
			ScheduledForDeletion: false,
			Config:               []byte(`{}`),
		}
		upsertConnector(t, ctx, store, connector4)

		// Create 2 accounts for connector3
		accounts3 := []models.Account{
			{
				ID: models.AccountID{
					Reference:   "isolation-test-1",
					ConnectorID: connector3.ID,
				},
				ConnectorID: connector3.ID,
				Reference:   "isolation-test-1",
				CreatedAt:   now.Add(-40 * time.Minute).UTC().Time,
				Type:        models.ACCOUNT_TYPE_INTERNAL,
				Raw:         []byte(`{}`),
			},
			{
				ID: models.AccountID{
					Reference:   "isolation-test-2",
					ConnectorID: connector3.ID,
				},
				ConnectorID: connector3.ID,
				Reference:   "isolation-test-2",
				CreatedAt:   now.Add(-40 * time.Minute).UTC().Time,
				Type:        models.ACCOUNT_TYPE_INTERNAL,
				Raw:         []byte(`{}`),
			},
		}
		upsertAccounts(t, ctx, store, accounts3)

		// Create 1 account for connector4
		accounts4 := []models.Account{
			{
				ID: models.AccountID{
					Reference:   "isolation-test-3",
					ConnectorID: connector4.ID,
				},
				ConnectorID: connector4.ID,
				Reference:   "isolation-test-3",
				CreatedAt:   now.Add(-40 * time.Minute).UTC().Time,
				Type:        models.ACCOUNT_TYPE_INTERNAL,
				Raw:         []byte(`{}`),
			},
		}
		upsertAccounts(t, ctx, store, accounts4)

		// Delete from connector3
		rowsAffected, err := store.AccountsDeleteFromConnectorIDBatch(ctx, connector3.ID, 10)
		require.NoError(t, err)
		require.Equal(t, 2, rowsAffected)

		// Verify connector3 accounts are deleted
		for _, a := range accounts3 {
			_, err := store.AccountsGet(ctx, a.ID)
			require.Error(t, err)
			require.ErrorIs(t, err, ErrNotFound)
		}

		// Verify connector4 account still exists
		account, err := store.AccountsGet(ctx, accounts4[0].ID)
		require.NoError(t, err)
		require.Equal(t, accounts4[0].ID, account.ID)
	})
}