HIP-1056 Add token blockitem to recorditem transformers

docs/design/block-streams.md

@@ -407,13 +407,7 @@ Beginning from an `EventTransaction` block item, a record item is composed of on
 | token_airdrop.token_id (fungible)  | state_changes[i].state_change.map_update.key.tokenReference.fungible_token_type                   |
 | token_airdrop.token_id (nft)       | state_changes[i].state_change.map_update.key.tokenReference.non_fungible_token.nftId.tokenId      |
 | assessed_custom_fee                | Similar to crypto_transfer, retrieved from transaction_output.token_airdrop                       |
-| token_account.account_id           | transaction_output.token_airdrop.automatic_token_associations[i].accountId                        |
-| token_account.token_id             | transaction_output.token_airdrop.automatic_token_associations[i].tokenId                          |
-| token_transfer.account_id          | transaction_output.token_airdrop.token_transfer_lists[i].transfers[j].accountAmount.accountID     |
-| token_transfer.amount              | transaction_output.token_airdrop.token_transfer_lists[i].transfers[j].accountAmount.amount        |
 | token_transfer.consensus_timestamp | transaction_result.consensus_timestamp                                                            |
-| token_transfer.is_approval         | transaction_output.token_airdrop.token_transfer_lists[i].transfers[j].accountAmount.isApproval    |
-| token_transfer.token_id            | transaction_output.token_airdrop.token_transfer_lists[i].token.tokenID                            |
 
 ### Token Create Transaction
 

hedera-mirror-importer/src/main/java/com/hedera/mirror/importer/downloader/block/transformer/AbstractBlockItemTransformer.java

@@ -53,11 +53,12 @@ public TransactionRecord getTransactionRecord(BlockItem blockItem, TransactionBo
             transactionRecordBuilder.setScheduleRef(transactionResult.getScheduleRef());
         }
 
-        updateTransactionRecord(blockItem, transactionRecordBuilder);
+        updateTransactionRecord(blockItem, transactionBody, transactionRecordBuilder);
         return transactionRecordBuilder.build();
     }
 
-    protected void updateTransactionRecord(BlockItem blockItem, TransactionRecord.Builder transactionRecordBuilder) {
+    protected void updateTransactionRecord(
+            BlockItem blockItem, TransactionBody transactionBody, TransactionRecord.Builder transactionRecordBuilder) {
         // do nothing
     }
 

hedera-mirror-importer/src/main/java/com/hedera/mirror/importer/downloader/block/transformer/AbstractTokenTransformer.java

@@ -0,0 +1,55 @@
+/*
+ * Copyright (C) 2025 Hedera Hashgraph, LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.hedera.mirror.importer.downloader.block.transformer;
+
+import static com.hedera.hapi.block.stream.output.protoc.StateIdentifier.STATE_ID_TOKENS;
+
+import com.hedera.hapi.block.stream.output.protoc.StateChange;
+import com.hedera.mirror.common.domain.transaction.BlockItem;
+import com.hederahashgraph.api.proto.java.TransactionRecord;
+import jakarta.inject.Named;
+
+@Named
+abstract class AbstractTokenTransformer extends AbstractBlockItemTransformer {
+
+    void updateTotalSupply(BlockItem blockItem, TransactionRecord.Builder transactionRecordBuilder) {
+        if (!blockItem.successful()) {
+            return;
+        }
+
+        for (var stateChanges : blockItem.stateChanges()) {
+            for (var stateChange : stateChanges.getStateChangesList()) {
+                if (hasSupplyUpdate(stateChange)) {
+                    updateTotalSupply(stateChange, transactionRecordBuilder);
+                    return;
+                }
+            }
+        }
+    }
+
+    boolean hasSupplyUpdate(StateChange stateChange) {
+        return stateChange.getStateId() == STATE_ID_TOKENS.getNumber()
+                && stateChange.hasMapUpdate()
+                && stateChange.getMapUpdate().hasValue()
+                && stateChange.getMapUpdate().getValue().hasTokenValue();
+    }
+
+    void updateTotalSupply(StateChange stateChange, TransactionRecord.Builder transactionRecordBuilder) {
+        var value = stateChange.getMapUpdate().getValue().getTokenValue().getTotalSupply();
+        transactionRecordBuilder.getReceiptBuilder().setNewTotalSupply(value);
+    }
+}

hedera-mirror-importer/src/main/java/com/hedera/mirror/importer/downloader/block/transformer/ConsensusCreateTopicTransformer.java

@@ -19,13 +19,15 @@
 import com.hedera.hapi.block.stream.output.protoc.StateIdentifier;
 import com.hedera.mirror.common.domain.transaction.BlockItem;
 import com.hedera.mirror.common.domain.transaction.TransactionType;
+import com.hederahashgraph.api.proto.java.TransactionBody;
 import com.hederahashgraph.api.proto.java.TransactionRecord;
 import jakarta.inject.Named;
 
 @Named
 final class ConsensusCreateTopicTransformer extends AbstractBlockItemTransformer {
     @Override
-    protected void updateTransactionRecord(BlockItem blockItem, TransactionRecord.Builder transactionRecordBuilder) {
+    protected void updateTransactionRecord(
+            BlockItem blockItem, TransactionBody transactionBody, TransactionRecord.Builder transactionRecordBuilder) {
 
         if (!blockItem.successful()) {
             return;

hedera-mirror-importer/src/main/java/com/hedera/mirror/importer/downloader/block/transformer/ConsensusSubmitMessageTransformer.java

@@ -21,14 +21,16 @@
 import com.hedera.hapi.block.stream.output.protoc.StateIdentifier;
 import com.hedera.mirror.common.domain.transaction.BlockItem;
 import com.hedera.mirror.common.domain.transaction.TransactionType;
+import com.hederahashgraph.api.proto.java.TransactionBody;
 import com.hederahashgraph.api.proto.java.TransactionRecord;
 import jakarta.inject.Named;
 
 @Named
 final class ConsensusSubmitMessageTransformer extends AbstractBlockItemTransformer {
 
     @Override
-    protected void updateTransactionRecord(BlockItem blockItem, TransactionRecord.Builder transactionRecordBuilder) {
+    protected void updateTransactionRecord(
+            BlockItem blockItem, TransactionBody transactionBody, TransactionRecord.Builder transactionRecordBuilder) {
 
         if (!blockItem.successful()) {
             return;

hedera-mirror-importer/src/main/java/com/hedera/mirror/importer/downloader/block/transformer/CryptoTransferTransformer.java

@@ -18,14 +18,16 @@
 
 import com.hedera.mirror.common.domain.transaction.BlockItem;
 import com.hedera.mirror.common.domain.transaction.TransactionType;
+import com.hederahashgraph.api.proto.java.TransactionBody;
 import com.hederahashgraph.api.proto.java.TransactionRecord;
 import jakarta.inject.Named;
 
 @Named
 final class CryptoTransferTransformer extends AbstractBlockItemTransformer {
 
     @Override
-    protected void updateTransactionRecord(BlockItem blockItem, TransactionRecord.Builder transactionRecordBuilder) {
+    protected void updateTransactionRecord(
+            BlockItem blockItem, TransactionBody transactionBody, TransactionRecord.Builder transactionRecordBuilder) {
         for (var transactionOutput : blockItem.transactionOutput()) {
             if (transactionOutput.hasCryptoTransfer()) {
                 var cryptoTransferOutput = transactionOutput.getCryptoTransfer();

hedera-mirror-importer/src/main/java/com/hedera/mirror/importer/downloader/block/transformer/FileCreateTransformer.java

@@ -19,14 +19,16 @@
 import com.hedera.hapi.block.stream.output.protoc.StateIdentifier;
 import com.hedera.mirror.common.domain.transaction.BlockItem;
 import com.hedera.mirror.common.domain.transaction.TransactionType;
+import com.hederahashgraph.api.proto.java.TransactionBody;
 import com.hederahashgraph.api.proto.java.TransactionRecord;
 import jakarta.inject.Named;
 
 @Named
 final class FileCreateTransformer extends AbstractBlockItemTransformer {
 
     @Override
-    protected void updateTransactionRecord(BlockItem blockItem, TransactionRecord.Builder transactionRecordBuilder) {
+    protected void updateTransactionRecord(
+            BlockItem blockItem, TransactionBody transactionBody, TransactionRecord.Builder transactionRecordBuilder) {
 
         if (!blockItem.successful()) {
             return;

hedera-mirror-importer/src/main/java/com/hedera/mirror/importer/downloader/block/transformer/NodeCreateTransformer.java

@@ -19,14 +19,16 @@
 import com.hedera.hapi.block.stream.output.protoc.StateIdentifier;
 import com.hedera.mirror.common.domain.transaction.BlockItem;
 import com.hedera.mirror.common.domain.transaction.TransactionType;
+import com.hederahashgraph.api.proto.java.TransactionBody;
 import com.hederahashgraph.api.proto.java.TransactionRecord;
 import jakarta.inject.Named;
 
 @Named
 final class NodeCreateTransformer extends AbstractBlockItemTransformer {
 
     @Override
-    protected void updateTransactionRecord(BlockItem blockItem, TransactionRecord.Builder transactionRecordBuilder) {
+    protected void updateTransactionRecord(
+            BlockItem blockItem, TransactionBody transactionBody, TransactionRecord.Builder transactionRecordBuilder) {
         if (!blockItem.successful()) {
             return;
         }

hedera-mirror-importer/src/main/java/com/hedera/mirror/importer/downloader/block/transformer/ScheduleCreateTransformer.java

@@ -20,6 +20,7 @@
 
 import com.hedera.mirror.common.domain.transaction.BlockItem;
 import com.hedera.mirror.common.domain.transaction.TransactionType;
+import com.hederahashgraph.api.proto.java.TransactionBody;
 import com.hederahashgraph.api.proto.java.TransactionRecord;
 import jakarta.inject.Named;
 
@@ -28,7 +29,8 @@
 final class ScheduleCreateTransformer extends AbstractBlockItemTransformer {
 
     @Override
-    protected void updateTransactionRecord(BlockItem blockItem, TransactionRecord.Builder transactionRecordBuilder) {
+    protected void updateTransactionRecord(
+            BlockItem blockItem, TransactionBody transactionBody, TransactionRecord.Builder transactionRecordBuilder) {
         if (!blockItem.successful()) {
             return;
         }

hedera-mirror-importer/src/main/java/com/hedera/mirror/importer/downloader/block/transformer/ScheduleSignTransformer.java

@@ -18,14 +18,16 @@
 
 import com.hedera.mirror.common.domain.transaction.BlockItem;
 import com.hedera.mirror.common.domain.transaction.TransactionType;
+import com.hederahashgraph.api.proto.java.TransactionBody;
 import com.hederahashgraph.api.proto.java.TransactionRecord;
 import jakarta.inject.Named;
 
 @Named
 final class ScheduleSignTransformer extends AbstractBlockItemTransformer {
 
     @Override
-    protected void updateTransactionRecord(BlockItem blockItem, TransactionRecord.Builder transactionRecordBuilder) {
+    protected void updateTransactionRecord(
+            BlockItem blockItem, TransactionBody transactionBody, TransactionRecord.Builder transactionRecordBuilder) {
         if (!blockItem.successful()) {
             return;
         }

hedera-mirror-importer/src/main/java/com/hedera/mirror/importer/downloader/block/transformer/TokenAirdropTransformer.java

@@ -0,0 +1,158 @@
+/*
+ * Copyright (C) 2025 Hedera Hashgraph, LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.hedera.mirror.importer.downloader.block.transformer;
+
+import static com.hedera.hapi.block.stream.output.protoc.StateIdentifier.STATE_ID_PENDING_AIRDROPS;
+
+import com.hedera.hapi.block.stream.output.protoc.StateChanges;
+import com.hedera.mirror.common.domain.transaction.BlockItem;
+import com.hedera.mirror.common.domain.transaction.TransactionType;
+import com.hederahashgraph.api.proto.java.AccountAmount;
+import com.hederahashgraph.api.proto.java.AccountID;
+import com.hederahashgraph.api.proto.java.NftID;
+import com.hederahashgraph.api.proto.java.NftTransfer;
+import com.hederahashgraph.api.proto.java.PendingAirdropId;
+import com.hederahashgraph.api.proto.java.PendingAirdropId.TokenReferenceCase;
+import com.hederahashgraph.api.proto.java.PendingAirdropRecord;
+import com.hederahashgraph.api.proto.java.TokenID;
+import com.hederahashgraph.api.proto.java.TokenTransferList;
+import com.hederahashgraph.api.proto.java.TransactionBody;
+import com.hederahashgraph.api.proto.java.TransactionRecord;
+import jakarta.inject.Named;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+@Named
+final class TokenAirdropTransformer extends AbstractBlockItemTransformer {
+
+    @Override
+    protected void updateTransactionRecord(
+            BlockItem blockItem, TransactionBody transactionBody, TransactionRecord.Builder transactionRecordBuilder) {
+        if (!blockItem.successful() || !transactionBody.hasTokenAirdrop()) {
+            return;
+        }
+
+        var pendingAirdropIds = pendingAirdropsInState(blockItem.stateChanges());
+        if (!pendingAirdropIds.isEmpty()) {
+            var eligibleAirdropIds =
+                    eligibleAirdropIds(transactionBody.getTokenAirdrop().getTokenTransfersList());
+            for (var pendingAirdrop : pendingAirdropIds) {
+                // Do not add airdrops that could not appear in the transfer list
+                if (eligibleAirdropIds.contains(pendingAirdrop.getPendingAirdropId())) {
+                    transactionRecordBuilder.addNewPendingAirdrops(pendingAirdrop);
+                }
+            }
+        }
+
+        for (var transactionOutput : blockItem.transactionOutput()) {
+            if (transactionOutput.hasTokenAirdrop()) {
+                var output = transactionOutput.getTokenAirdrop();
+                var assessedCustomFees = output.getAssessedCustomFeesList();
+                transactionRecordBuilder.addAllAssessedCustomFees(assessedCustomFees);
+            }
+        }
+    }
+
+    private Set<PendingAirdropRecord> pendingAirdropsInState(List<StateChanges> stateChangesList) {
+        Set<PendingAirdropRecord> pendingAirdropIds = new HashSet<>();
+        for (var stateChanges : stateChangesList) {
+            for (var stateChange : stateChanges.getStateChangesList()) {
+                if (stateChange.getStateId() == STATE_ID_PENDING_AIRDROPS.getNumber()
+                        && stateChange.hasMapUpdate()
+                        && stateChange.getMapUpdate().hasKey()
+                        && stateChange.getMapUpdate().getKey().hasPendingAirdropIdKey()) {
+                    var mapUpdate = stateChange.getMapUpdate();
+                    var pendingId = mapUpdate.getKey().getPendingAirdropIdKey();
+                    var pendingAirdrop = PendingAirdropRecord.newBuilder().setPendingAirdropId(pendingId);
+                    if (pendingId.getTokenReferenceCase() == TokenReferenceCase.FUNGIBLE_TOKEN_TYPE
+                            && mapUpdate.getValue().hasAccountPendingAirdropValue()
+                            && mapUpdate
+                                    .getValue()
+                                    .getAccountPendingAirdropValue()
+                                    .hasPendingAirdropValue()) {
+                        var accountValue = mapUpdate.getValue().getAccountPendingAirdropValue();
+                        pendingAirdrop.setPendingAirdropValue(accountValue.getPendingAirdropValue());
+                    }
+
+                    pendingAirdropIds.add(pendingAirdrop.build());
+                }
+            }
+        }
+
+        return pendingAirdropIds;
+    }
+
+    private Set<PendingAirdropId> eligibleAirdropIds(List<TokenTransferList> tokenTransfers) {
+        var eligibleAirdrops = new HashSet<PendingAirdropId>();
+        for (var transfer : tokenTransfers) {
+            var tokenId = transfer.getToken();
+            var accountAmounts = transfer.getTransfersList();
+            if (!accountAmounts.isEmpty()) {
+                eligibleFungiblePendingAirdrops(accountAmounts, tokenId, eligibleAirdrops);
+            }
+
+            var nftTransfers = transfer.getNftTransfersList();
+            if (!nftTransfers.isEmpty()) {
+                eligibleNftPendingAirdrops(nftTransfers, tokenId, eligibleAirdrops);
+            }
+        }
+
+        return eligibleAirdrops;
+    }
+
+    private void eligibleFungiblePendingAirdrops(
+            List<AccountAmount> accountAmounts, TokenID tokenId, Set<PendingAirdropId> eligibleAirdrops) {
+        var builder = PendingAirdropId.newBuilder().setFungibleTokenType(tokenId);
+        var receivers = new HashSet<AccountID>();
+        var senders = new HashSet<AccountID>();
+        for (var accountAmount : accountAmounts) {
+            if (accountAmount.hasAccountID()) {
+                var accountId = accountAmount.getAccountID();
+                if (accountAmount.getAmount() < 0) {
+                    senders.add(accountId);
+                } else {
+                    receivers.add(accountId);
+                }
+            }
+        }
+
+        for (var receiver : receivers) {
+            for (var sender : senders) {
+                eligibleAirdrops.add(
+                        builder.setReceiverId(receiver).setSenderId(sender).build());
+            }
+        }
+    }
+
+    private void eligibleNftPendingAirdrops(
+            List<NftTransfer> nftTransfers, TokenID tokenId, Set<PendingAirdropId> eligibleAirdrops) {
+        for (var nftTransfer : nftTransfers) {
+            var nftId = NftID.newBuilder().setTokenID(tokenId).setSerialNumber(nftTransfer.getSerialNumber());
+            eligibleAirdrops.add(PendingAirdropId.newBuilder()
+                    .setNonFungibleToken(nftId)
+                    .setReceiverId(nftTransfer.getReceiverAccountID())
+                    .setSenderId(nftTransfer.getSenderAccountID())
+                    .build());
+        }
+    }
+
+    @Override
+    public TransactionType getType() {
+        return TransactionType.TOKENAIRDROP;
+    }
+}