chore: batch insert miner_sector_deal (#1221)

Author: birdychang

chain/indexer/integrated/processor/state.go

@@ -469,9 +469,14 @@ func MakeProcessors(api tasks.DataSource, indexerTasks []string) (*IndexerProces
 				minertask.PreCommitInfoExtractorV9{},
 			)
 		case tasktype.MinerSectorDeal:
-			out.ActorProcessors[t] = actorstate.NewTask(api, actorstate.NewTypedActorExtractorMap(
-				mineractors.AllCodes(), minertask.SectorDealsExtractor{},
-			))
+			out.ActorProcessors[t] = actorstate.NewTaskWithTransformer(
+				api,
+				actorstate.NewTypedActorExtractorMap(
+					mineractors.AllCodes(),
+					minertask.SectorDealsExtractor{},
+				),
+				minertask.SectorDealsExtractor{},
+			)
 		case tasktype.MinerSectorEvent:
 			out.ActorProcessors[t] = actorstate.NewTaskWithTransformer(
 				api,

chain/indexer/integrated/processor/state_internal_test.go

@@ -81,7 +81,7 @@ func TestNewProcessor(t *testing.T) {
 	require.Equal(t, actorstate.NewTask(nil, actorstate.NewTypedActorExtractorMap(miner.AllCodes(), minertask.FeeDebtExtractor{})), proc.actorProcessors[tasktype.MinerFeeDebt])
 	require.Equal(t, actorstate.NewTask(nil, actorstate.NewTypedActorExtractorMap(miner.AllCodes(), minertask.InfoExtractor{})), proc.actorProcessors[tasktype.MinerInfo])
 	require.Equal(t, actorstate.NewTaskWithTransformer(nil, actorstate.NewTypedActorExtractorMap(miner.AllCodes(), minertask.LockedFundsExtractor{}), minertask.LockedFundsExtractor{}), proc.actorProcessors[tasktype.MinerLockedFund])
-	require.Equal(t, actorstate.NewTask(nil, actorstate.NewTypedActorExtractorMap(miner.AllCodes(), minertask.SectorDealsExtractor{})), proc.actorProcessors[tasktype.MinerSectorDeal])
+	require.Equal(t, actorstate.NewTaskWithTransformer(nil, actorstate.NewTypedActorExtractorMap(miner.AllCodes(), minertask.SectorDealsExtractor{}), minertask.SectorDealsExtractor{}), proc.actorProcessors[tasktype.MinerSectorDeal])
 	require.Equal(t, actorstate.NewTaskWithTransformer(nil, actorstate.NewTypedActorExtractorMap(miner.AllCodes(), minertask.SectorEventsExtractor{}), minertask.SectorEventsExtractor{}), proc.actorProcessors[tasktype.MinerSectorEvent])
 	require.Equal(t, actorstate.NewTask(nil, actorstate.NewTypedActorExtractorMap(miner.AllCodes(), minertask.PoStExtractor{})), proc.actorProcessors[tasktype.MinerSectorPost])
 	require.Equal(t, actorstate.NewTask(nil, actorstate.NewCustomTypedActorExtractorMap(

chain/indexer/integrated/processor/state_test.go

@@ -62,10 +62,6 @@ func TestMakeProcessorsActors(t *testing.T) {
 				taskName:  tasktype.MinerInfo,
 				extractor: actorstate.NewTypedActorExtractorMap(miner.AllCodes(), minertask.InfoExtractor{}),
 			},
-			{
-				taskName:  tasktype.MinerSectorDeal,
-				extractor: actorstate.NewTypedActorExtractorMap(miner.AllCodes(), minertask.SectorDealsExtractor{}),
-			},
 			{
 				taskName:  tasktype.MinerSectorPost,
 				extractor: actorstate.NewTypedActorExtractorMap(miner.AllCodes(), minertask.PoStExtractor{}),
@@ -139,6 +135,11 @@ func TestMakeProcessorsActors(t *testing.T) {
 				),
 				transformer: minertask.PreCommitInfoExtractorV9{},
 			},
+			{
+				taskName:    tasktype.MinerSectorDeal,
+				extractor:   actorstate.NewTypedActorExtractorMap(miner.AllCodes(), minertask.SectorDealsExtractor{}),
+				transformer: minertask.SectorDealsExtractor{},
+			},
 		}
 		for _, tc := range testCases2 {
 			t.Run(tc.taskName, func(t *testing.T) {

tasks/actorstate/miner/sector_deals.go

@@ -72,3 +72,17 @@ func ExtractSectorDealsModel(ctx context.Context, ec extraction.State) (minermod
 	}
 	return result, nil
 }
+
+func (SectorDealsExtractor) Transform(ctx context.Context, data model.PersistableList) (model.PersistableList, error) {
+	persistableList := make(minermodel.MinerSectorDealList, 0, len(data))
+	for _, d := range data {
+		ml, ok := d.(minermodel.MinerSectorDealList)
+		if !ok {
+			return nil, fmt.Errorf("expected MinerSectorDealList type but got: %T", d)
+		}
+		for _, m := range ml {
+			persistableList = append(persistableList, m)
+		}
+	}
+	return model.PersistableList{persistableList}, nil
+}