Allow synchronous 200 response for token pool creation

db/migrations/postgres/000058_remove_tokenpool_key.down.sql

@@ -0,0 +1,4 @@
+BEGIN;
+ALTER TABLE tokenpool ADD COLUMN "key" VARCHAR(1024);
+UPDATE tokenpool SET "key" = '';
+COMMIT;

db/migrations/postgres/000058_remove_tokenpool_key.up.sql

@@ -0,0 +1,3 @@
+BEGIN;
+ALTER TABLE tokenpool DROP COLUMN key;
+COMMIT;

db/migrations/sqlite/000058_remove_tokenpool_key.down.sql

@@ -0,0 +1,2 @@
+ALTER TABLE tokenpool ADD COLUMN "key" VARCHAR(1024);
+UPDATE tokenpool SET "key" = '';

db/migrations/sqlite/000058_remove_tokenpool_key.up.sql

@@ -0,0 +1 @@
+ALTER TABLE tokenpool DROP COLUMN key;

docs/swagger/swagger.yaml

@@ -6838,11 +6838,6 @@ paths:
         name: id
         schema:
           type: string
-      - description: 'Data filter field. Prefixes supported: > >= < <= @ ^ ! !@ !^'
-        in: query
-        name: key
-        schema:
-          type: string
       - description: 'Data filter field. Prefixes supported: > >= < <= @ ^ ! !@ !^'
         in: query
         name: message
@@ -9008,11 +9003,6 @@ paths:
         name: id
         schema:
           type: string
-      - description: 'Data filter field. Prefixes supported: > >= < <= @ ^ ! !@ !^'
-        in: query
-        name: key
-        schema:
-          type: string
       - description: 'Data filter field. Prefixes supported: > >= < <= @ ^ ! !@ !^'
         in: query
         name: message

internal/database/sqlcommon/tokenpool_sql.go

@@ -1,4 +1,4 @@
-// Copyright © 2021 Kaleido, Inc.
+// Copyright © 2022 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -42,7 +42,6 @@ var (
 		"created",
 		"tx_type",
 		"tx_id",
-		"key",
 	}
 	tokenPoolFilterFieldMap = map[string]string{
 		"protocolid":       "protocol_id",
@@ -94,7 +93,6 @@ func (s *SQLCommon) UpsertTokenPool(ctx context.Context, pool *fftypes.TokenPool
 				Set("state", pool.State).
 				Set("tx_type", pool.TX.Type).
 				Set("tx_id", pool.TX.ID).
-				Set("key", pool.Key).
 				Where(sq.Eq{"id": pool.ID}),
 			func() {
 				s.callbacks.UUIDCollectionNSEvent(database.CollectionTokenPools, fftypes.ChangeEventTypeUpdated, pool.Namespace, pool.ID)
@@ -121,7 +119,6 @@ func (s *SQLCommon) UpsertTokenPool(ctx context.Context, pool *fftypes.TokenPool
 					pool.Created,
 					pool.TX.Type,
 					pool.TX.ID,
-					pool.Key,
 				),
 			func() {
 				s.callbacks.UUIDCollectionNSEvent(database.CollectionTokenPools, fftypes.ChangeEventTypeCreated, pool.Namespace, pool.ID)
@@ -150,7 +147,6 @@ func (s *SQLCommon) tokenPoolResult(ctx context.Context, row *sql.Rows) (*fftype
 		&pool.Created,
 		&pool.TX.Type,
 		&pool.TX.ID,
-		&pool.Key,
 	)
 	if err != nil {
 		return nil, i18n.WrapError(ctx, err, i18n.MsgDBReadErr, "tokenpool")

internal/database/sqlcommon/tokenpool_sql_test.go

@@ -1,4 +1,4 @@
-// Copyright © 2021 Kaleido, Inc.
+// Copyright © 2022 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -52,7 +52,6 @@ func TestTokenPoolE2EWithDB(t *testing.T) {
 			Type: fftypes.TransactionTypeTokenPool,
 			ID:   fftypes.NewUUID(),
 		},
-		Key: "0x12345",
 	}
 
 	s.callbacks.On("UUIDCollectionNSEvent", database.CollectionTokenPools, fftypes.ChangeEventTypeCreated, "ns1", poolID, mock.Anything).

internal/events/token_pool_created.go

@@ -30,7 +30,6 @@ import (
 func addPoolDetailsFromPlugin(ffPool *fftypes.TokenPool, pluginPool *tokens.TokenPool) {
 	ffPool.Type = pluginPool.Type
 	ffPool.ProtocolID = pluginPool.ProtocolID
-	ffPool.Key = pluginPool.Key
 	ffPool.Connector = pluginPool.Connector
 	ffPool.Standard = pluginPool.Standard
 	if pluginPool.TransactionID != nil {
@@ -64,7 +63,7 @@ func (em *eventManager) confirmPool(ctx context.Context, pool *fftypes.TokenPool
 	if err := em.database.UpsertTokenPool(ctx, pool); err != nil {
 		return err
 	}
-	log.L(ctx).Infof("Token pool confirmed id=%s author=%s", pool.ID, pool.Key)
+	log.L(ctx).Infof("Token pool confirmed, id=%s", pool.ID)
 	event := fftypes.NewEvent(fftypes.EventTypePoolConfirmed, pool.Namespace, pool.ID)
 	return em.database.InsertEvent(ctx, event)
 }
@@ -133,20 +132,17 @@ func (em *eventManager) shouldAnnounce(ctx context.Context, ti tokens.Plugin, po
 // It will be invoked on every node (including the submitter) after the pool is announced+activated, to trigger confirmation of the pool.
 // When received in any other scenario, it should be ignored.
 func (em *eventManager) TokenPoolCreated(ti tokens.Plugin, pool *tokens.TokenPool) (err error) {
-	if pool.TransactionID == nil {
-		log.L(em.ctx).Errorf("Invalid token pool transaction - ID is nil")
-		return nil // move on
-	}
-
 	var batchID *fftypes.UUID
 	var announcePool *fftypes.TokenPool
 
 	err = em.retry.Do(em.ctx, "persist token pool transaction", func(attempt int) (bool, error) {
 		err := em.database.RunAsGroup(em.ctx, func(ctx context.Context) error {
 			// See if this is a confirmation of an unconfirmed pool
-			if existingPool, err := em.shouldConfirm(ctx, pool); err != nil {
+			existingPool, err := em.shouldConfirm(ctx, pool)
+			if err != nil {
 				return err
-			} else if existingPool != nil {
+			}
+			if existingPool != nil {
 				if existingPool.State == fftypes.TokenPoolStateConfirmed {
 					return nil // already confirmed
 				}
@@ -156,6 +152,11 @@ func (em *eventManager) TokenPoolCreated(ti tokens.Plugin, pool *tokens.TokenPoo
 					batchID = msg.BatchID // trigger rewind after completion of database transaction
 				}
 				return em.confirmPool(ctx, existingPool, &pool.Event, pool.Event.BlockchainTXID)
+			} else if pool.TransactionID == nil {
+				// TransactionID is required if the pool doesn't exist yet
+				// (but it may be omitted when activating a pool that was received via definition broadcast)
+				log.L(em.ctx).Errorf("Invalid token pool transaction - ID is nil")
+				return nil // move on
 			}
 
 			// See if this pool was submitted locally and needs to be announced
@@ -186,7 +187,7 @@ func (em *eventManager) TokenPoolCreated(ti tokens.Plugin, pool *tokens.TokenPoo
 				Pool:  announcePool,
 				Event: buildBlockchainEvent(announcePool.Namespace, nil, &pool.Event, &announcePool.TX),
 			}
-			log.L(em.ctx).Infof("Announcing token pool id=%s author=%s", announcePool.ID, pool.Key)
+			log.L(em.ctx).Infof("Announcing token pool, id=%s", announcePool.ID)
 			_, err = em.broadcast.BroadcastTokenPool(em.ctx, announcePool.Namespace, broadcast, false)
 		}
 	}

internal/events/token_pool_created_test.go

@@ -43,7 +43,6 @@ func TestTokenPoolCreatedIgnore(t *testing.T) {
 	pool := &tokens.TokenPool{
 		Type:          fftypes.TokenTypeFungible,
 		ProtocolID:    "123",
-		Key:           "0x0",
 		TransactionID: txID,
 		Connector:     "erc1155",
 		Event: blockchain.Event{
@@ -65,13 +64,13 @@ func TestTokenPoolCreatedIgnore(t *testing.T) {
 func TestTokenPoolCreatedIgnoreNoTX(t *testing.T) {
 	em, cancel := newTestEventManager(t)
 	defer cancel()
+	mdi := em.database.(*databasemocks.Plugin)
 	mti := &tokenmocks.Plugin{}
 
 	info := fftypes.JSONObject{"some": "info"}
 	pool := &tokens.TokenPool{
 		Type:          fftypes.TokenTypeFungible,
 		ProtocolID:    "123",
-		Key:           "0x0",
 		TransactionID: nil,
 		Connector:     "erc1155",
 		Event: blockchain.Event{
@@ -81,6 +80,8 @@ func TestTokenPoolCreatedIgnoreNoTX(t *testing.T) {
 		},
 	}
 
+	mdi.On("GetTokenPoolByProtocolID", em.ctx, "erc1155", "123").Return(nil, nil, nil)
+
 	err := em.TokenPoolCreated(mti, pool)
 	assert.NoError(t, err)
 }
@@ -96,7 +97,6 @@ func TestTokenPoolCreatedConfirm(t *testing.T) {
 	chainPool := &tokens.TokenPool{
 		Type:          fftypes.TokenTypeFungible,
 		ProtocolID:    "123",
-		Key:           "0x0",
 		Connector:     "erc1155",
 		TransactionID: txID,
 		Event: blockchain.Event{
@@ -109,7 +109,6 @@ func TestTokenPoolCreatedConfirm(t *testing.T) {
 	storedPool := &fftypes.TokenPool{
 		Namespace: "ns1",
 		ID:        fftypes.NewUUID(),
-		Key:       chainPool.Key,
 		State:     fftypes.TokenPoolStatePending,
 		Message:   fftypes.NewUUID(),
 		TX: fftypes.TransactionRef{
@@ -156,7 +155,6 @@ func TestTokenPoolCreatedAlreadyConfirmed(t *testing.T) {
 	chainPool := &tokens.TokenPool{
 		Type:          fftypes.TokenTypeFungible,
 		ProtocolID:    "123",
-		Key:           "0x0",
 		Connector:     "erc1155",
 		TransactionID: txID,
 		Event: blockchain.Event{
@@ -168,7 +166,6 @@ func TestTokenPoolCreatedAlreadyConfirmed(t *testing.T) {
 	storedPool := &fftypes.TokenPool{
 		Namespace: "ns1",
 		ID:        fftypes.NewUUID(),
-		Key:       chainPool.Key,
 		State:     fftypes.TokenPoolStateConfirmed,
 		TX: fftypes.TransactionRef{
 			Type: fftypes.TransactionTypeTokenPool,
@@ -196,7 +193,6 @@ func TestTokenPoolCreatedMigrate(t *testing.T) {
 	chainPool := &tokens.TokenPool{
 		Type:          fftypes.TokenTypeFungible,
 		ProtocolID:    "123",
-		Key:           "0x0",
 		Connector:     "magic-tokens",
 		TransactionID: txID,
 		Event: blockchain.Event{
@@ -208,7 +204,6 @@ func TestTokenPoolCreatedMigrate(t *testing.T) {
 	storedPool := &fftypes.TokenPool{
 		Namespace: "ns1",
 		ID:        fftypes.NewUUID(),
-		Key:       chainPool.Key,
 		State:     fftypes.TokenPoolStateUnknown,
 		TX: fftypes.TransactionRef{
 			Type: fftypes.TransactionTypeTokenPool,
@@ -374,7 +369,6 @@ func TestTokenPoolCreatedAnnounce(t *testing.T) {
 	pool := &tokens.TokenPool{
 		Type:          fftypes.TokenTypeFungible,
 		ProtocolID:    "123",
-		Key:           "0x0",
 		TransactionID: txID,
 		Connector:     "erc1155",
 		Event: blockchain.Event{
@@ -421,7 +415,6 @@ func TestTokenPoolCreatedAnnounceBadOpInputID(t *testing.T) {
 	pool := &tokens.TokenPool{
 		Type:          fftypes.TokenTypeFungible,
 		ProtocolID:    "123",
-		Key:           "0x0",
 		TransactionID: txID,
 		Connector:     "erc1155",
 		Event: blockchain.Event{
@@ -460,7 +453,6 @@ func TestTokenPoolCreatedAnnounceBadOpInputNS(t *testing.T) {
 	pool := &tokens.TokenPool{
 		Type:          fftypes.TokenTypeFungible,
 		ProtocolID:    "123",
-		Key:           "0x0",
 		TransactionID: txID,
 		Connector:     "erc1155",
 		Event: blockchain.Event{

internal/tokens/fftokens/fftokens.go

@@ -177,11 +177,10 @@ func (ft *FFTokens) handleTokenPoolCreate(ctx context.Context, data fftypes.JSON
 	eventProtocolID := data.GetString("id")
 	tokenType := data.GetString("type")
 	protocolID := data.GetString("poolId")
-	standard := data.GetString("standard") // this is optional
-	operatorAddress := data.GetString("operator")
-	tx := data.GetObject("transaction")
-	txHash := tx.GetString("transactionHash")
+	standard := data.GetString("standard")   // optional
 	rawOutput := data.GetObject("rawOutput") // optional
+	tx := data.GetObject("transaction")
+	txHash := tx.GetString("transactionHash") // optional
 
 	timestampStr := data.GetString("timestamp")
 	timestamp, err := fftypes.ParseTimeString(timestampStr)
@@ -190,9 +189,7 @@ func (ft *FFTokens) handleTokenPoolCreate(ctx context.Context, data fftypes.JSON
 	}
 
 	if tokenType == "" ||
-		protocolID == "" ||
-		operatorAddress == "" ||
-		txHash == "" {
+		protocolID == "" {
 		log.L(ctx).Errorf("TokenPool event is not valid - missing data: %+v", data)
 		return nil // move on
 	}
@@ -210,7 +207,6 @@ func (ft *FFTokens) handleTokenPoolCreate(ctx context.Context, data fftypes.JSON
 		Type:          fftypes.FFEnum(tokenType),
 		ProtocolID:    protocolID,
 		TransactionID: poolData.TX,
-		Key:           operatorAddress,
 		Connector:     ft.configuredName,
 		Standard:      standard,
 		Event: blockchain.Event{
@@ -235,11 +231,11 @@ func (ft *FFTokens) handleTokenTransfer(ctx context.Context, t fftypes.TokenTran
 	fromAddress := data.GetString("from")
 	toAddress := data.GetString("to")
 	value := data.GetString("amount")
-	tx := data.GetObject("transaction")
-	txHash := tx.GetString("transactionHash")
 	tokenIndex := data.GetString("tokenIndex") // optional
 	uri := data.GetString("uri")               // optional
 	rawOutput := data.GetObject("rawOutput")   // optional
+	tx := data.GetObject("transaction")
+	txHash := tx.GetString("transactionHash") // optional
 
 	timestampStr := data.GetString("timestamp")
 	timestamp, err := fftypes.ParseTimeString(timestampStr)
@@ -261,7 +257,6 @@ func (ft *FFTokens) handleTokenTransfer(ctx context.Context, t fftypes.TokenTran
 		poolProtocolID == "" ||
 		operatorAddress == "" ||
 		value == "" ||
-		txHash == "" ||
 		(t != fftypes.TokenTransferTypeMint && fromAddress == "") ||
 		(t != fftypes.TokenTransferTypeBurn && toAddress == "") {
 		log.L(ctx).Errorf("%s event is not valid - missing data: %+v", eventName, data)
@@ -392,6 +387,14 @@ func (ft *FFTokens) CreateTokenPool(ctx context.Context, opID *fftypes.UUID, poo
 	if err != nil || !res.IsSuccess() {
 		return restclient.WrapRestErr(ctx, res, err, i18n.MsgTokensRESTErr)
 	}
+	if res.StatusCode() == 200 {
+		// Handle synchronous response (202 will be handled by later websocket listener)
+		var obj fftypes.JSONObject
+		if err := json.Unmarshal(res.Body(), &obj); err != nil {
+			return i18n.WrapError(ctx, err, i18n.MsgJSONObjectParseFailed, res.Body())
+		}
+		return ft.handleTokenPoolCreate(ctx, obj)
+	}
 	return nil
 }
 
@@ -406,6 +409,14 @@ func (ft *FFTokens) ActivateTokenPool(ctx context.Context, opID *fftypes.UUID, p
 	if err != nil || !res.IsSuccess() {
 		return restclient.WrapRestErr(ctx, res, err, i18n.MsgTokensRESTErr)
 	}
+	if res.StatusCode() == 200 {
+		// Handle synchronous response (202 will be handled by later websocket listener)
+		var obj fftypes.JSONObject
+		if err := json.Unmarshal(res.Body(), &obj); err != nil {
+			return i18n.WrapError(ctx, err, i18n.MsgJSONObjectParseFailed, res.Body())
+		}
+		return ft.handleTokenPoolCreate(ctx, obj)
+	}
 	return nil
 }