feat(payments/transferinitiations): store reference + list by reference

components/payments/cmd/api/internal/storage/transfer_initiation.go

@@ -14,7 +14,7 @@ func (s *Storage) GetTransferInitiation(ctx context.Context, id models.TransferI
 	var transferInitiation models.TransferInitiation
 
 	query := s.db.NewSelect().
-		Column("id", "connector_id", "created_at", "scheduled_at", "description", "type", "source_account_id", "destination_account_id", "provider", "initial_amount", "amount", "asset", "metadata").
+		Column("id", "reference", "connector_id", "created_at", "scheduled_at", "description", "type", "source_account_id", "destination_account_id", "provider", "initial_amount", "amount", "asset", "metadata").
 		Model(&transferInitiation).
 		Relation("RelatedAdjustments").
 		Where("id = ?", id)
@@ -68,7 +68,7 @@ func (s *Storage) ListTransferInitiations(ctx context.Context, q ListTransferIni
 		(*bunpaginate.OffsetPaginatedQuery[PaginatedQueryOptions[TransferInitiationQuery]])(&q),
 		func(query *bun.SelectQuery) *bun.SelectQuery {
 			query = query.
-				Column("id", "connector_id", "created_at", "scheduled_at", "description", "type", "source_account_id", "destination_account_id", "provider", "initial_amount", "amount", "asset", "metadata").
+				Column("id", "reference", "connector_id", "created_at", "scheduled_at", "description", "type", "source_account_id", "destination_account_id", "provider", "initial_amount", "amount", "asset", "metadata").
 				Relation("RelatedAdjustments")
 
 			if q.Options.QueryBuilder != nil {
@@ -105,6 +105,17 @@ func (s *Storage) transferInitiationQueryContext(qb query.Builder) (string, []an
 			default:
 				return "", nil, fmt.Errorf("unexpected type %T for column '%s'", accountID, key)
 			}
+		case key == "reference":
+			if operator != "$match" {
+				return "", nil, fmt.Errorf("'%s' column can only be used with $match", key)
+			}
+
+			switch reference := value.(type) {
+			case string:
+				return fmt.Sprintf("%s = ?", key), []any{reference}, nil
+			default:
+				return "", nil, fmt.Errorf("unexpected type %T for column '%s'", reference, key)
+			}
 		default:
 			return "", nil, fmt.Errorf("unknown key '%s' when building query", key)
 		}

components/payments/cmd/connectors/internal/api/service/transfer_initiation.go

@@ -140,6 +140,7 @@ func (s *Service) CreateTransferInitiation(ctx context.Context, req *CreateTrans
 	createdAt := time.Now().UTC()
 	tf := &models.TransferInitiation{
 		ID:                   id,
+		Reference:            req.Reference,
 		CreatedAt:            createdAt,
 		ScheduledAt:          req.ScheduledAt,
 		Description:          req.Description,

components/payments/cmd/connectors/internal/storage/transfer_initiation.go

@@ -19,7 +19,7 @@ func (s *Storage) CreateTransferInitiation(ctx context.Context, transferInitiati
 	}()
 
 	query := tx.NewInsert().
-		Column("id", "created_at", "scheduled_at", "description", "type", "destination_account_id", "provider", "connector_id", "initial_amount", "amount", "asset", "metadata").
+		Column("id", "reference", "created_at", "scheduled_at", "description", "type", "destination_account_id", "provider", "connector_id", "initial_amount", "amount", "asset", "metadata").
 		Model(transferInitiation)
 
 	if transferInitiation.SourceAccountID != nil {
@@ -53,7 +53,7 @@ func (s *Storage) ReadTransferInitiation(ctx context.Context, id models.Transfer
 	var transferInitiation models.TransferInitiation
 
 	query := s.db.NewSelect().
-		Column("id", "created_at", "scheduled_at", "description", "type", "source_account_id", "destination_account_id", "provider", "connector_id", "amount", "asset", "metadata").
+		Column("id", "reference", "created_at", "scheduled_at", "description", "type", "source_account_id", "destination_account_id", "provider", "connector_id", "amount", "asset", "metadata").
 		Model(&transferInitiation).
 		Relation("RelatedAdjustments").
 		Where("id = ?", id)

components/payments/internal/models/transfer_initiation.go

@@ -115,6 +115,7 @@ type TransferInitiation struct {
 	// Filled when created in DB
 	ID TransferInitiationID `bun:",pk,nullzero"`
 
+	Reference   string
 	CreatedAt   time.Time `bun:",nullzero"`
 	ScheduledAt time.Time `bun:",nullzero"`
 	Description string

components/payments/internal/storage/migrations_v1.x.go

@@ -391,6 +391,11 @@ func registerMigrationsV1(ctx context.Context, migrator *migrations.Migrator) {
 				return fixExpandingChangelogs(ctx, tx)
 			},
 		},
+		migrations.Migration{
+			Up: func(tx bun.Tx) error {
+				return fixMissingReferenceTransferInitiation(ctx, tx)
+			},
+		},
 	)
 }
 
@@ -1090,3 +1095,54 @@ func fixExpandingChangelogs(ctx context.Context, tx bun.Tx) error {
 
 	return nil
 }
+
+func fixMissingReferenceTransferInitiation(ctx context.Context, tx bun.Tx) error {
+	_, err := tx.Exec(`
+		ALTER TABLE transfers.transfer_initiation ADD COLUMN IF NOT EXISTS reference text;
+	`)
+	if err != nil {
+		return err
+	}
+
+	var createdAt time.Time
+	for {
+		var transferInitiations []models.TransferInitiation
+		query := tx.NewSelect().
+			Model(&transferInitiations).
+			Order("created_at ASC").
+			Limit(100)
+
+		if !createdAt.IsZero() {
+			query.Where("created_at > ?", createdAt)
+		}
+
+		err := query.Scan(ctx)
+		if err != nil {
+			return err
+		}
+
+		fmt.Println("Processing", len(transferInitiations), "transfer initiations")
+		if len(transferInitiations) == 0 {
+			break
+		}
+
+		for i := range transferInitiations {
+			if transferInitiations[i].Reference == "" {
+				transferInitiations[i].Reference = transferInitiations[i].ID.Reference
+			}
+			createdAt = transferInitiations[i].CreatedAt
+		}
+
+		fmt.Println("Updating", len(transferInitiations), "transfer initiations")
+		_, err = tx.NewInsert().
+			Model(&transferInitiations).
+			On("CONFLICT (id) DO UPDATE").
+			Set("reference = EXCLUDED.reference").
+			Exec(ctx)
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}