payments/internal/storage/open_banking_connections.go

package storage

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	libtime "time"

	"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"
	internalEvents "github.com/hanzo-fi/payments/internal/events"
	"github.com/hanzo-fi/payments/pkg/domain/models"
	"github.com/hanzo-fi/payments/pkg/events"
	"github.com/google/uuid"
	"github.com/uptrace/bun"
)

type openBankingConnectionAttempt struct {
	bun.BaseModel `bun:"table:open_banking_connection_attempts"`

	// Mandatory fields
	ID          uuid.UUID                                 `bun:"id,pk,type:uuid,notnull"`
	PsuID       uuid.UUID                                 `bun:"psu_id,type:uuid,notnull"`
	ConnectorID models.ConnectorID                        `bun:"connector_id,type:character varying,notnull"`
	CreatedAt   time.Time                                 `bun:"created_at,type:timestamp without time zone,notnull"`
	Status      models.OpenBankingConnectionAttemptStatus `bun:"status,type:text,notnull"`
	State       json.RawMessage                           `bun:"state,type:jsonb,nullzero"`

	// Optional fields
	ClientRedirectURL *string    `bun:"client_redirect_url,type:text,nullzero"`
	ExpiresAt         *time.Time `bun:"expires_at,type:timestamp without time zone,nullzero"`
	Error             *string    `bun:"error,type:text,nullzero"`

	// Encrypted field: written via pgp_sym_encrypt and scanned back decrypted.
	// The raw temporary_token column is bytea; this field receives the decrypted value.
	TemporaryToken *string `bun:"decrypted_temporary_token,scanonly"`
}

func (s *store) OpenBankingConnectionAttemptsUpsert(ctx context.Context, from models.OpenBankingConnectionAttempt) error {
	attempt, err := fromOpenBankingConnectionAttemptsModels(from)
	if err != nil {
		return err
	}

	_, err = s.db.NewInsert().
		Model(&attempt).
		// temporary_token is encrypted at rest (bytea); NULL stays NULL.
		Value("temporary_token", "CASE WHEN ?::TEXT IS NULL THEN NULL ELSE pgp_sym_encrypt(?::TEXT, ?, ?) END",
			attempt.TemporaryToken, attempt.TemporaryToken, s.configEncryptionKey, encryptionOptions).
		On("CONFLICT (id) DO UPDATE").
		Set("error = EXCLUDED.error").
		Set("status = EXCLUDED.status").
		Set("temporary_token = EXCLUDED.temporary_token").
		Set("expires_at = EXCLUDED.expires_at").
		Set("state = EXCLUDED.state").
		Exec(ctx)
	if err != nil {
		return e("upserting open banking connection attempt", err)
	}

	return nil
}

func (s *store) OpenBankingConnectionAttemptsUpdateStatus(ctx context.Context, id uuid.UUID, status models.OpenBankingConnectionAttemptStatus, errMsg *string) error {
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return e("failed to begin transaction", err)
	}
	defer func() {
		rollbackOnTxError(ctx, &tx, err)
	}()

	_, err = tx.NewUpdate().
		Model((*openBankingConnectionAttempt)(nil)).
		Set("status = ?", status).
		Set("error = ?", errMsg).
		Where("id = ?", id).
		Exec(ctx)
	if err != nil {
		return e("updating open banking connection attempt status", err)
	}

	// Fetch the attempt to get psuID and connectorID for the event.
	// Select columns explicitly (like the get/list paths) so the query never
	// references the raw encrypted temporary_token column; the event payload
	// does not need the token, so it is intentionally omitted.
	attempt := openBankingConnectionAttempt{}
	err = tx.NewSelect().
		Model(&attempt).
		Column("id", "psu_id", "connector_id", "created_at", "status", "state", "client_redirect_url", "expires_at", "error").
		Where("id = ?", id).
		Scan(ctx)
	if err != nil {
		return e("failed to get open banking connection attempt", err)
	}

	var attemptModel *models.OpenBankingConnectionAttempt
	attemptModel, err = toOpenBankingConnectionAttemptsModels(attempt)
	if err != nil {
		return e("failed to convert open banking connection attempt", err)
	}

	// Create outbox event for user link status
	payload := internalEvents.OpenBankingUserLinkStatus{
		PsuID:       attemptModel.PsuID,
		ConnectorID: attemptModel.ConnectorID.String(),
		Status:      string(status),
		Error:       errMsg,
	}

	var payloadBytes []byte
	payloadBytes, err = json.Marshal(payload)
	if err != nil {
		return e("failed to marshal user link status event payload", err)
	}

	idempotencyKey := models.IdempotencyKey(payload)
	outboxEvent := models.OutboxEvent{
		ID: models.EventID{
			EventIdempotencyKey: idempotencyKey,
			ConnectorID:         &attemptModel.ConnectorID,
		},
		EventType:   events.EventTypeOpenBankingUserLinkStatus,
		EntityID:    attemptModel.PsuID.String(),
		Payload:     payloadBytes,
		CreatedAt:   libtime.Now().UTC(),
		Status:      models.OUTBOX_STATUS_PENDING,
		ConnectorID: &attemptModel.ConnectorID,
	}

	if err = s.OutboxEventsInsert(ctx, tx, []models.OutboxEvent{outboxEvent}); err != nil {
		return err
	}

	if err = tx.Commit(); err != nil {
		return e("failed to commit transaction", err)
	}
	return nil
}

func (s *store) OpenBankingConnectionAttemptsGet(ctx context.Context, id uuid.UUID) (*models.OpenBankingConnectionAttempt, error) {
	attempt := openBankingConnectionAttempt{}
	err := s.db.NewSelect().
		Model(&attempt).
		Column("id", "psu_id", "connector_id", "created_at", "status", "state", "client_redirect_url", "expires_at", "error").
		ColumnExpr("pgp_sym_decrypt(temporary_token, ?, ?) AS decrypted_temporary_token", s.configEncryptionKey, encryptionOptions).
		Where("id = ?", id).
		Scan(ctx)
	if err != nil {
		return nil, e("getting open banking connection attempt", err)
	}

	return toOpenBankingConnectionAttemptsModels(attempt)
}

type OpenBankingConnectionAttemptsQuery struct{}

type ListOpenBankingConnectionAttemptsQuery paginate.OffsetPaginatedQuery[paginate.PaginatedQueryOptions[OpenBankingConnectionAttemptsQuery]]

func NewListOpenBankingConnectionAttemptsQuery(opts paginate.PaginatedQueryOptions[OpenBankingConnectionAttemptsQuery]) ListOpenBankingConnectionAttemptsQuery {
	return ListOpenBankingConnectionAttemptsQuery{
		PageSize: opts.PageSize,
		Order:    paginate.OrderAsc,
		Options:  opts,
	}
}

func (s *store) openBankingConnectionAttemptsQueryContext(qb query.Builder) (string, []any, error) {
	return qb.Build(query.ContextFn(func(key, operator string, value any) (string, []any, error) {
		switch key {
		case "id":
			if operator != "$match" {
				return "", nil, fmt.Errorf("'%s' column can only be used with $match: %w", key, ErrValidation)
			}
			return fmt.Sprintf("%s = ?", key), []any{value}, nil
		case "status":
			if operator != "$match" {
				return "", nil, fmt.Errorf("'%s' column can only be used with $match: %w", key, ErrValidation)
			}
			return fmt.Sprintf("%s = ?", key), []any{value}, nil
		default:
			return "", nil, fmt.Errorf("unknown key '%s' when building query: %w", key, ErrValidation)
		}
	}))
}

func (s *store) OpenBankingConnectionAttemptsList(ctx context.Context, psuID uuid.UUID, connectorID models.ConnectorID, query ListOpenBankingConnectionAttemptsQuery) (*paginate.Cursor[models.OpenBankingConnectionAttempt], error) {
	var (
		where string
		args  []any
		err   error
	)
	if query.Options.QueryBuilder != nil {
		where, args, err = s.openBankingConnectionAttemptsQueryContext(query.Options.QueryBuilder)
		if err != nil {
			return nil, err
		}
	}

	cursor, err := paginateWithOffset[paginate.PaginatedQueryOptions[OpenBankingConnectionAttemptsQuery], openBankingConnectionAttempt](s, ctx,
		(*paginate.OffsetPaginatedQuery[paginate.PaginatedQueryOptions[OpenBankingConnectionAttemptsQuery]])(&query),
		func(query *bun.SelectQuery) *bun.SelectQuery {
			query = query.
				Column("id", "psu_id", "connector_id", "created_at", "status", "state", "client_redirect_url", "expires_at", "error").
				ColumnExpr("pgp_sym_decrypt(temporary_token, ?, ?) AS decrypted_temporary_token", s.configEncryptionKey, encryptionOptions)

			if where != "" {
				query = query.Where(where, args...)
			}

			query = query.Where("psu_id = ?", psuID)
			query = query.Where("connector_id = ?", connectorID)

			query = query.Order("created_at DESC")

			return query
		},
	)
	if err != nil {
		return nil, e("failed to fetch open banking connection attempts", err)
	}

	openBankingConnectionAttemptsModels := make([]models.OpenBankingConnectionAttempt, len(cursor.Data))
	for i, attempt := range cursor.Data {
		res, err := toOpenBankingConnectionAttemptsModels(attempt)
		if err != nil {
			return nil, e("failed to fetch open banking connection attempts", err)
		}
		openBankingConnectionAttemptsModels[i] = *res
	}

	return &paginate.Cursor[models.OpenBankingConnectionAttempt]{
		PageSize: cursor.PageSize,
		HasMore:  cursor.HasMore,
		Previous: cursor.Previous,
		Next:     cursor.Next,
		Data:     openBankingConnectionAttemptsModels,
	}, nil
}

type openBankingForwardedUser struct {
	bun.BaseModel `bun:"table:open_banking_forwarded_users,alias:open_banking_forwarded_users"`

	// Mandatory fields
	PsuID       uuid.UUID          `bun:"psu_id,pk,type:uuid,notnull"`
	ConnectorID models.ConnectorID `bun:"connector_id,pk,type:character varying,notnull"`

	// Optional fields
	PSPUserID *string           `bun:"psp_user_id,type:text,nullzero"`
	Metadata  map[string]string `bun:"metadata,type:jsonb,nullzero,notnull,default:'{}'"`

	// Scan only fields
	AccessToken *string    `bun:"access_token,type:text,nullzero,scanonly"`
	ExpiresAt   *time.Time `bun:"expires_at,type:timestamp without time zone,nullzero,scanonly"`
}

func (s *store) OpenBankingForwardedUserUpsert(ctx context.Context, psuID uuid.UUID, from models.OpenBankingForwardedUser) error {
	openBankingForwardedUser, token := fromOpenBankingForwardedUserModels(from, psuID)

	tx, err := s.db.BeginTx(ctx, &sql.TxOptions{})
	if err != nil {
		return e("failed to create transaction", err)
	}
	defer func() {
		rollbackOnTxError(ctx, &tx, err)
	}()

	if token != nil {
		_, err = tx.NewInsert().
			Model(token).
			// access_token is encrypted at rest (bytea).
			Value("access_token", "pgp_sym_encrypt(?::TEXT, ?, ?)", token.AccessToken, s.configEncryptionKey, encryptionOptions).
			On("CONFLICT (psu_id, connector_id, connection_id) DO UPDATE").
			Set("access_token = EXCLUDED.access_token").
			Set("expires_at = EXCLUDED.expires_at").
			Exec(ctx)
		if err != nil {
			return e("upserting open banking forwarded user access token", err)
		}
	}

	_, err = tx.NewInsert().
		Model(&openBankingForwardedUser).
		On("CONFLICT (psu_id, connector_id) DO UPDATE").
		Set("psp_user_id = EXCLUDED.psp_user_id").
		Set("expires_at = EXCLUDED.expires_at").
		Set("metadata = EXCLUDED.metadata").
		Exec(ctx)
	if err != nil {
		return e("upserting open banking forwarded user", err)
	}

	if err = tx.Commit(); err != nil {
		return e("failed to commit transaction", err)
	}
	return nil
}

func (s *store) OpenBankingForwardedUserGet(ctx context.Context, psuID uuid.UUID, connectorID models.ConnectorID) (*models.OpenBankingForwardedUser, error) {
	obForwardedUser := openBankingForwardedUser{}
	err := s.db.NewSelect().
		Model(&obForwardedUser).
		Column("open_banking_forwarded_users.*", "open_banking_access_tokens.expires_at").
		ColumnExpr("pgp_sym_decrypt(open_banking_access_tokens.access_token, ?, ?) AS access_token", s.configEncryptionKey, encryptionOptions).
		Join("LEFT JOIN open_banking_access_tokens ON open_banking_forwarded_users.psu_id = open_banking_access_tokens.psu_id AND open_banking_forwarded_users.connector_id = open_banking_access_tokens.connector_id").
		Where("open_banking_forwarded_users.psu_id = ?", psuID).
		Where("open_banking_forwarded_users.connector_id = ?", connectorID).
		Scan(ctx)

	if err != nil {
		return nil, e("getting open banking forwarded user", err)
	}

	return toOpenBankingForwardedUserModels(obForwardedUser), nil
}

func (s *store) OpenBankingForwardedUserGetByPSPUserID(ctx context.Context, pspUserID string, connectorID models.ConnectorID) (*models.OpenBankingForwardedUser, error) {
	obForwardedUser := openBankingForwardedUser{}

	err := s.db.NewSelect().
		Model(&obForwardedUser).
		Column("open_banking_forwarded_users.*", "open_banking_access_tokens.expires_at").
		ColumnExpr("pgp_sym_decrypt(open_banking_access_tokens.access_token, ?, ?) AS access_token", s.configEncryptionKey, encryptionOptions).
		Join("LEFT JOIN open_banking_access_tokens ON open_banking_forwarded_users.psu_id = open_banking_access_tokens.psu_id AND open_banking_forwarded_users.connector_id = open_banking_access_tokens.connector_id").
		Where("open_banking_forwarded_users.psp_user_id = ?", pspUserID).
		Where("open_banking_forwarded_users.connector_id = ?", connectorID).
		Scan(ctx)

	if err != nil {
		return nil, e("getting open banking forwarded user", err)
	}

	return toOpenBankingForwardedUserModels(obForwardedUser), nil
}

func (s *store) OpenBankingForwardedUserDelete(ctx context.Context, psuID uuid.UUID, connectorID models.ConnectorID) error {
	_, err := s.db.NewDelete().
		Model((*openBankingForwardedUser)(nil)).
		Where("psu_id = ?", psuID).
		Where("connector_id = ?", connectorID).
		Exec(ctx)
	if err != nil {
		return e("deleting open banking forwarded user", err)
	}

	return nil
}

type OpenBankingForwardedUserQuery struct{}

type ListOpenBankingForwardedUserQuery paginate.OffsetPaginatedQuery[paginate.PaginatedQueryOptions[OpenBankingForwardedUserQuery]]

func NewListOpenBankingForwardedUserQuery(opts paginate.PaginatedQueryOptions[OpenBankingForwardedUserQuery]) ListOpenBankingForwardedUserQuery {
	return ListOpenBankingForwardedUserQuery{
		Order:    paginate.OrderAsc,
		PageSize: opts.PageSize,
		Options:  opts,
	}
}

func (s *store) openBankingForwardedUserQueryContext(qb query.Builder) (string, []any, error) {
	return qb.Build(query.ContextFn(func(key, operator string, value any) (string, []any, error) {
		switch {
		case key == "connector_id", key == "psu_id":
			if operator != "$match" {
				return "", nil, fmt.Errorf("'%s' column can only be used with $match: %w", key, ErrValidation)
			}
			return fmt.Sprintf("open_banking_forwarded_users.%s = ?", key), []any{value}, nil
		case metadataRegex.Match([]byte(key)):
			if operator != "$match" {
				return "", nil, fmt.Errorf("'metadata' column can only be used with $match: %w", ErrValidation)
			}
			match := metadataRegex.FindAllStringSubmatch(key, 3)

			key := "open_banking_forwarded_users.metadata"
			return key + " @> ?", []any{map[string]any{
				match[0][1]: value,
			}}, nil
		default:
			return "", nil, fmt.Errorf("unknown key '%s' when building query: %w", key, ErrValidation)
		}
	}))
}

func (s *store) OpenBankingForwardedUserList(ctx context.Context, query ListOpenBankingForwardedUserQuery) (*paginate.Cursor[models.OpenBankingForwardedUser], error) {
	var (
		where string
		args  []any
		err   error
	)
	if query.Options.QueryBuilder != nil {
		where, args, err = s.openBankingForwardedUserQueryContext(query.Options.QueryBuilder)
		if err != nil {
			return nil, err
		}
	}

	cursor, err := paginateWithOffset[paginate.PaginatedQueryOptions[OpenBankingForwardedUserQuery], openBankingForwardedUser](s, ctx,
		(*paginate.OffsetPaginatedQuery[paginate.PaginatedQueryOptions[OpenBankingForwardedUserQuery]])(&query),
		func(query *bun.SelectQuery) *bun.SelectQuery {
			if where != "" {
				query = query.Where(where, args...)
			}

			query = query.
				Column("open_banking_forwarded_users.*", "open_banking_access_tokens.expires_at").
				ColumnExpr("pgp_sym_decrypt(open_banking_access_tokens.access_token, ?, ?) AS access_token", s.configEncryptionKey, encryptionOptions).
				Join("LEFT JOIN open_banking_access_tokens ON open_banking_forwarded_users.psu_id = open_banking_access_tokens.psu_id AND open_banking_forwarded_users.connector_id = open_banking_access_tokens.connector_id")

			return query
		},
	)
	if err != nil {
		return nil, e("failed to fetch open banking forwarded user", err)
	}

	obForwardedUserModels := make([]models.OpenBankingForwardedUser, len(cursor.Data))
	for i, p := range cursor.Data {
		bb := toOpenBankingForwardedUserModels(p)
		obForwardedUserModels[i] = *bb
	}

	return &paginate.Cursor[models.OpenBankingForwardedUser]{
		PageSize: cursor.PageSize,
		HasMore:  cursor.HasMore,
		Previous: cursor.Previous,
		Next:     cursor.Next,
		Data:     obForwardedUserModels,
	}, nil
}

type openBankingConnections struct {
	bun.BaseModel `bun:"table:open_banking_connections"`

	// Mandatory fields
	PsuID         uuid.UUID               `bun:"psu_id,pk,type:uuid,notnull"`
	ConnectorID   models.ConnectorID      `bun:"connector_id,pk,type:character varying,notnull"`
	ConnectionID  string                  `bun:"connection_id,pk,type:character varying,notnull"`
	CreatedAt     time.Time               `bun:"created_at,type:timestamp without time zone,notnull"`
	DataUpdatedAt time.Time               `bun:"data_updated_at,type:timestamp without time zone,notnull"`
	Status        models.ConnectionStatus `bun:"status,type:text,notnull"`
	UpdatedAt     time.Time               `bun:"updated_at,type:timestamp without time zone,notnull"`

	// Optional fields
	Error    *string           `bun:"error,type:text,nullzero"`
	Metadata map[string]string `bun:"metadata,type:jsonb,nullzero,notnull,default:'{}'"`

	// ScanOnly fields
	AccessToken *string    `bun:"access_token,type:text,nullzero,scanonly"`
	ExpiresAt   *time.Time `bun:"expires_at,type:timestamp without time zone,nullzero,scanonly"`
}

func (s *store) OpenBankingConnectionsUpsert(ctx context.Context, psuID uuid.UUID, from models.OpenBankingConnection) error {
	connection, token := fromOpenBankingConnectionsModels(from, psuID)

	tx, err := s.db.BeginTx(ctx, &sql.TxOptions{})
	if err != nil {
		return e("failed to create transaction", err)
	}
	defer func() {
		rollbackOnTxError(ctx, &tx, err)
	}()

	// Check if connection already exists and get its previous status to determine if this is a reconnect
	var existingConnection openBankingConnections
	var connectionExists bool
	err = tx.NewSelect().
		Model(&existingConnection).
		Where("psu_id = ?", psuID).
		Where("connector_id = ?", from.ConnectorID).
		Where("connection_id = ?", from.ConnectionID).
		Scan(ctx)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			connectionExists = false
		} else {
			return e("failed to check if connection exists", err)
		}
	} else {
		connectionExists = true
	}

	if token != nil {
		_, err = tx.NewInsert().
			Model(token).
			// access_token is encrypted at rest (bytea).
			Value("access_token", "pgp_sym_encrypt(?::TEXT, ?, ?)", token.AccessToken, s.configEncryptionKey, encryptionOptions).
			On("CONFLICT (psu_id, connector_id, connection_id) DO UPDATE").
			Set("access_token = EXCLUDED.access_token").
			Set("expires_at = EXCLUDED.expires_at").
			Exec(ctx)
		if err != nil {
			return e("upserting open banking connection access token", err)
		}
	}

	_, err = tx.NewInsert().
		Model(&connection).
		On("CONFLICT (psu_id, connector_id, connection_id) DO UPDATE").
		Set("metadata = EXCLUDED.metadata").
		Set("status = EXCLUDED.status").
		Set("error = EXCLUDED.error").
		Set("updated_at = EXCLUDED.updated_at").
		Exec(ctx)
	if err != nil {
		return e("upserting open banking connection", err)
	}

	// Create outbox events based on connection status
	var outboxEvents []models.OutboxEvent

	if from.Status == models.ConnectionStatusError {
		// Connection disconnected
		payload := internalEvents.OpenBankingUserConnectionDisconnected{
			PsuID:        psuID,
			ConnectorID:  from.ConnectorID.String(),
			ConnectionID: from.ConnectionID,
			ErrorType:    string(from.Status),
			At:           from.UpdatedAt,
			Reason:       from.Error,
		}

		var payloadBytes []byte
		payloadBytes, err = json.Marshal(payload)
		if err != nil {
			return e("failed to marshal user connection disconnected event payload", err)
		}

		idempotencyKey := models.IdempotencyKey(payload)
		outboxEvents = append(outboxEvents, models.OutboxEvent{
			ID: models.EventID{
				EventIdempotencyKey: idempotencyKey,
				ConnectorID:         &from.ConnectorID,
			},
			EventType:   events.EventTypeOpenBankingUserConnectionDisconnected,
			EntityID:    from.ConnectionID,
			Payload:     payloadBytes,
			CreatedAt:   libtime.Now().UTC(),
			Status:      models.OUTBOX_STATUS_PENDING,
			ConnectorID: &from.ConnectorID,
		})
	} else if from.Status == models.ConnectionStatusActive && connectionExists && existingConnection.Status == models.ConnectionStatusError {
		// Connection reconnected (was ERROR, now ACTIVE)
		payload := internalEvents.OpenBankingUserConnectionReconnected{
			PsuID:        psuID,
			ConnectorID:  from.ConnectorID.String(),
			ConnectionID: from.ConnectionID,
			At:           from.UpdatedAt,
		}

		var payloadBytes []byte
		payloadBytes, err = json.Marshal(payload)
		if err != nil {
			return e("failed to marshal user connection reconnected event payload", err)
		}

		idempotencyKey := models.IdempotencyKey(payload)
		outboxEvents = append(outboxEvents, models.OutboxEvent{
			ID: models.EventID{
				EventIdempotencyKey: idempotencyKey,
				ConnectorID:         &from.ConnectorID,
			},
			EventType:   events.EventTypeOpenBankingUserConnectionReconnected,
			EntityID:    from.ConnectionID,
			Payload:     payloadBytes,
			CreatedAt:   libtime.Now().UTC(),
			Status:      models.OUTBOX_STATUS_PENDING,
			ConnectorID: &from.ConnectorID,
		})
	}

	if len(outboxEvents) > 0 {
		if err = s.OutboxEventsInsert(ctx, tx, outboxEvents); err != nil {
			return err
		}
	}

	if err = tx.Commit(); err != nil {
		return e("failed to commit transaction", err)
	}
	return nil
}

func (s *store) OpenBankingConnectionsUpdateLastDataUpdate(ctx context.Context, psuID uuid.UUID, connectorID models.ConnectorID, connectionID string, updatedAt libtime.Time) error {
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return e("failed to begin transaction", err)
	}
	defer func() {
		rollbackOnTxError(ctx, &tx, err)
	}()

	_, err = tx.NewUpdate().
		Model((*openBankingConnections)(nil)).
		Set("data_updated_at = ?", updatedAt).
		Where("psu_id = ?", psuID).
		Where("connector_id = ?", connectorID).
		Where("connection_id = ?", connectionID).
		Exec(ctx)
	if err != nil {
		return e("updating open banking connection last data update", err)
	}

	// Create outbox event for user connection data synced
	payload := internalEvents.OpenBankingUserConnectionDataSynced{
		PsuID:        psuID,
		ConnectorID:  connectorID.String(),
		ConnectionID: connectionID,
		At:           updatedAt,
	}

	var payloadBytes []byte
	payloadBytes, err = json.Marshal(payload)
	if err != nil {
		return e("failed to marshal user connection data synced event payload", err)
	}

	idempotencyKey := models.IdempotencyKey(payload)
	outboxEvent := models.OutboxEvent{
		ID: models.EventID{
			EventIdempotencyKey: idempotencyKey,
			ConnectorID:         &connectorID,
		},
		EventType:   events.EventTypeOpenBankingUserConnectionDataSynced,
		EntityID:    connectionID,
		Payload:     payloadBytes,
		CreatedAt:   libtime.Now().UTC(),
		Status:      models.OUTBOX_STATUS_PENDING,
		ConnectorID: &connectorID,
	}

	if err = s.OutboxEventsInsert(ctx, tx, []models.OutboxEvent{outboxEvent}); err != nil {
		return err
	}

	if err = tx.Commit(); err != nil {
		return e("failed to commit transaction", err)
	}
	return nil
}

func (s *store) OpenBankingConnectionsGet(ctx context.Context, psuID uuid.UUID, connectorID models.ConnectorID, connectionID string) (*models.OpenBankingConnection, error) {
	connection := openBankingConnections{}
	err := s.db.NewSelect().
		Model(&connection).
		Column("open_banking_connections.*", "open_banking_access_tokens.expires_at").
		ColumnExpr("pgp_sym_decrypt(open_banking_access_tokens.access_token, ?, ?) AS access_token", s.configEncryptionKey, encryptionOptions).
		Join("LEFT JOIN open_banking_access_tokens ON open_banking_connections.psu_id = open_banking_access_tokens.psu_id AND open_banking_connections.connector_id = open_banking_access_tokens.connector_id AND open_banking_connections.connection_id = open_banking_access_tokens.connection_id").
		Where("open_banking_connections.psu_id = ?", psuID).
		Where("open_banking_connections.connector_id = ?", connectorID).
		Where("open_banking_connections.connection_id = ?", connectionID).
		Scan(ctx)
	if err != nil {
		return nil, e("getting open banking connection", err)
	}

	return pointer.For(toOpenBankingConnectionsModels(connection)), nil
}

func (s *store) OpenBankingConnectionsGetFromConnectionID(ctx context.Context, connectorID models.ConnectorID, connectionID string) (*models.OpenBankingConnection, uuid.UUID, error) {
	connection := openBankingConnections{}
	err := s.db.NewSelect().
		Model(&connection).
		Column("open_banking_connections.*", "open_banking_access_tokens.expires_at").
		ColumnExpr("pgp_sym_decrypt(open_banking_access_tokens.access_token, ?, ?) AS access_token", s.configEncryptionKey, encryptionOptions).
		Join("LEFT JOIN open_banking_access_tokens ON open_banking_connections.psu_id = open_banking_access_tokens.psu_id AND open_banking_connections.connector_id = open_banking_access_tokens.connector_id AND open_banking_connections.connection_id = open_banking_access_tokens.connection_id").
		Where("open_banking_connections.connector_id = ?", connectorID).
		Where("open_banking_connections.connection_id = ?", connectionID).
		Scan(ctx)
	if err != nil {
		return nil, uuid.Nil, e("getting open banking connection", err)
	}

	return pointer.For(toOpenBankingConnectionsModels(connection)), connection.PsuID, nil
}

func (s *store) OpenBankingConnectionsDelete(ctx context.Context, psuID uuid.UUID, connectorID models.ConnectorID, connectionID string) error {
	_, err := s.db.NewDelete().
		Model((*openBankingConnections)(nil)).
		Where("psu_id = ?", psuID).
		Where("connector_id = ?", connectorID).
		Where("connection_id = ?", connectionID).
		Exec(ctx)
	if err != nil {
		return e("deleting open banking connection", err)
	}

	return nil
}

type OpenBankingConnectionsQuery struct{}

type ListOpenBankingConnectionsQuery paginate.OffsetPaginatedQuery[paginate.PaginatedQueryOptions[OpenBankingConnectionsQuery]]

func NewListOpenBankingConnectionsQuery(opts paginate.PaginatedQueryOptions[OpenBankingConnectionsQuery]) ListOpenBankingConnectionsQuery {
	return ListOpenBankingConnectionsQuery{
		PageSize: opts.PageSize,
		Order:    paginate.OrderAsc,
		Options:  opts,
	}
}

func (s *store) openBankingConnectionsQueryContext(qb query.Builder) (string, []any, error) {
	return qb.Build(query.ContextFn(func(key, operator string, value any) (string, []any, error) {
		switch {
		case key == "connection_id", key == "status":
			if operator != "$match" {
				return "", nil, fmt.Errorf("'%s' column can only be used with $match: %w", key, ErrValidation)
			}
			return fmt.Sprintf("%s = ?", key), []any{value}, nil
		case metadataRegex.Match([]byte(key)):
			if operator != "$match" {
				return "", nil, fmt.Errorf("'metadata' column can only be used with $match: %w", ErrValidation)
			}
			match := metadataRegex.FindAllStringSubmatch(key, 3)

			key := "metadata"
			return key + " @> ?", []any{map[string]any{
				match[0][1]: value,
			}}, nil
		default:
			return "", nil, fmt.Errorf("unknown key '%s' when building query: %w", key, ErrValidation)
		}
	}))
}

func (s *store) OpenBankingConnectionsList(ctx context.Context, psuID uuid.UUID, connectorID *models.ConnectorID, query ListOpenBankingConnectionsQuery) (*paginate.Cursor[models.OpenBankingConnection], error) {
	var (
		where string
		args  []any
		err   error
	)
	if query.Options.QueryBuilder != nil {
		where, args, err = s.openBankingConnectionsQueryContext(query.Options.QueryBuilder)
		if err != nil {
			return nil, err
		}
	}

	cursor, err := paginateWithOffset[paginate.PaginatedQueryOptions[OpenBankingConnectionsQuery], openBankingConnections](s, ctx,
		(*paginate.OffsetPaginatedQuery[paginate.PaginatedQueryOptions[OpenBankingConnectionsQuery]])(&query),
		func(query *bun.SelectQuery) *bun.SelectQuery {
			if where != "" {
				query = query.Where(where, args...)
			}

			query = query.Where("psu_id = ?", psuID)

			if connectorID != nil {
				query = query.Where("connector_id = ?", connectorID)
			}

			query = query.Order("created_at DESC")

			return query
		},
	)
	if err != nil {
		return nil, e("failed to fetch open banking connections", err)
	}

	openBankingConnections := make([]openBankingConnections, len(cursor.Data))
	copy(openBankingConnections, cursor.Data)

	openBankingConnectionsModels := make([]models.OpenBankingConnection, len(openBankingConnections))
	for i, connection := range openBankingConnections {
		openBankingConnectionsModels[i] = toOpenBankingConnectionsModels(connection)
	}

	return &paginate.Cursor[models.OpenBankingConnection]{
		PageSize: cursor.PageSize,
		HasMore:  cursor.HasMore,
		Previous: cursor.Previous,
		Next:     cursor.Next,
		Data:     openBankingConnectionsModels,
	}, nil
}

type openBankingAccessTokens struct {
	bun.BaseModel `bun:"table:open_banking_access_tokens"`

	// Mandatory fields
	PSUID        uuid.UUID          `bun:"psu_id,type:uuid,notnull"`
	ConnectorID  models.ConnectorID `bun:"connector_id,type:character varying,notnull"`
	ConnectionID *string            `bun:"connection_id,type:character varying,nullzero"`
	AccessToken  string             `bun:"access_token,type:text,notnull"`
	ExpiresAt    time.Time          `bun:"expires_at,type:timestamp without time zone,notnull"`
}

func fromOpenBankingConnectionAttemptsModels(from models.OpenBankingConnectionAttempt) (openBankingConnectionAttempt, error) {
	var token *string
	var expiresAt *time.Time
	if from.TemporaryToken != nil {
		t, e := fromTokenModels(*from.TemporaryToken)
		token = &t
		expiresAt = &e
	}

	state, err := json.Marshal(from.State)
	if err != nil {
		return openBankingConnectionAttempt{}, err
	}

	return openBankingConnectionAttempt{
		ID:                from.ID,
		PsuID:             from.PsuID,
		ConnectorID:       from.ConnectorID,
		CreatedAt:         time.New(from.CreatedAt),
		Status:            from.Status,
		State:             state,
		ClientRedirectURL: from.ClientRedirectURL,
		TemporaryToken:    token,
		ExpiresAt:         expiresAt,
		Error:             from.Error,
	}, nil
}

func toOpenBankingConnectionAttemptsModels(from openBankingConnectionAttempt) (*models.OpenBankingConnectionAttempt, error) {
	state := models.CallbackState{}
	if err := json.Unmarshal(from.State, &state); err != nil {
		return nil, err
	}

	return &models.OpenBankingConnectionAttempt{
		ID:                from.ID,
		PsuID:             from.PsuID,
		ConnectorID:       from.ConnectorID,
		CreatedAt:         from.CreatedAt.Time,
		Status:            from.Status,
		State:             state,
		ClientRedirectURL: from.ClientRedirectURL,
		TemporaryToken:    toTokenModels(from.TemporaryToken, from.ExpiresAt),
		Error:             from.Error,
	}, nil
}

func fromOpenBankingForwardedUserModels(from models.OpenBankingForwardedUser, psuID uuid.UUID) (openBankingForwardedUser, *openBankingAccessTokens) {
	var token *openBankingAccessTokens
	if from.AccessToken != nil {
		accessToken, expiresAt := fromTokenModels(*from.AccessToken)

		token = &openBankingAccessTokens{
			PSUID:       psuID,
			ConnectorID: from.ConnectorID,
			AccessToken: accessToken,
			ExpiresAt:   expiresAt,
		}
	}

	return openBankingForwardedUser{
		PsuID:       psuID,
		PSPUserID:   from.PSPUserID,
		ConnectorID: from.ConnectorID,
		Metadata:    from.Metadata,
	}, token
}

func toOpenBankingForwardedUserModels(from openBankingForwardedUser) *models.OpenBankingForwardedUser {
	return &models.OpenBankingForwardedUser{
		PsuID:       from.PsuID,
		ConnectorID: from.ConnectorID,
		PSPUserID:   from.PSPUserID,
		AccessToken: toTokenModels(from.AccessToken, from.ExpiresAt),
		Metadata:    from.Metadata,
	}
}

func fromOpenBankingConnectionsModels(from models.OpenBankingConnection, psuID uuid.UUID) (openBankingConnections, *openBankingAccessTokens) {
	var token *openBankingAccessTokens
	if from.AccessToken != nil {
		accessToken, expiresAt := fromTokenModels(*from.AccessToken)

		token = &openBankingAccessTokens{
			PSUID:        psuID,
			ConnectorID:  from.ConnectorID,
			ConnectionID: &from.ConnectionID,
			AccessToken:  accessToken,
			ExpiresAt:    expiresAt,
		}
	}

	return openBankingConnections{
		PsuID:         psuID,
		ConnectorID:   from.ConnectorID,
		ConnectionID:  from.ConnectionID,
		CreatedAt:     time.New(from.CreatedAt),
		DataUpdatedAt: time.New(from.DataUpdatedAt),
		Status:        from.Status,
		UpdatedAt:     time.New(from.UpdatedAt),
		Error:         from.Error,
		Metadata:      from.Metadata,
	}, token
}

func toOpenBankingConnectionsModels(from openBankingConnections) models.OpenBankingConnection {
	return models.OpenBankingConnection{
		ConnectorID:   from.ConnectorID,
		ConnectionID:  from.ConnectionID,
		CreatedAt:     from.CreatedAt.Time,
		DataUpdatedAt: from.DataUpdatedAt.Time,
		Status:        from.Status,
		UpdatedAt:     from.UpdatedAt.Time,
		Error:         from.Error,
		AccessToken:   toTokenModels(from.AccessToken, from.ExpiresAt),
		Metadata:      from.Metadata,
	}
}

func fromTokenModels(from models.Token) (string, time.Time) {
	return from.Token, time.New(from.ExpiresAt)
}

func toTokenModels(from *string, expiresAt *time.Time) *models.Token {
	if from == nil {
		return nil
	}

	token := &models.Token{
		Token: *from,
	}

	if expiresAt != nil {
		token.ExpiresAt = expiresAt.Time
	}

	return token
}