Actually fix owner changes properly

service/pipeline/parsing_stage.go

@@ -14,6 +14,7 @@ import (
 	"github.com/aws/aws-sdk-go-v2/service/lambda"
 
 	"github.com/NFT-com/indexer/config/params"
+	"github.com/NFT-com/indexer/models/events"
 	"github.com/NFT-com/indexer/models/graph"
 	"github.com/NFT-com/indexer/models/jobs"
 	"github.com/NFT-com/indexer/models/results"
@@ -135,8 +136,16 @@ func (p *ParsingStage) process(payload []byte) error {
 	// We can go through the transfers and process those with zero address as mints.
 	var dummies []*graph.NFT
 	var payloads [][]byte
+	var owners []*events.Transfer
 	for _, transfer := range result.Transfers {
 
+		if transfer.SenderAddress == transfer.ReceiverAddress {
+			continue
+		}
+
+		// Collect transfers that are not no-ops for owner changes.
+		owners = append(owners, transfer)
+
 		// Get the collection ID based on chain ID and collection address, so we can
 		// reference it directly for the addition job and the NFT insertion.
 		collection, err := p.collections.One(transfer.ChainID, transfer.CollectionAddress)
@@ -199,7 +208,7 @@ func (p *ParsingStage) process(payload []byte) error {
 	}
 
 	// Last but not least, we can upsert the owner change updates for each transfer.
-	err = p.owners.Upsert(result.Transfers...)
+	err = p.owners.Upsert(owners...)
 	if err != nil {
 		return fmt.Errorf("could not upsert owners: %w", err)
 	}

storage/graph/owner_repository.go

@@ -37,7 +37,7 @@ func (n *OwnerRepository) Upsert(transfers ...*events.Transfer) error {
 			"event_id",
 			"number",
 		).
-		Suffix("ON CONFLICT (owner, nft_id, event_id) DO UPDATE SET number = owners.number + EXCLUDED.number")
+		Suffix("ON CONFLICT DO NOTHING")
 
 	for _, transfer := range transfers {