Migrate to Indexer SDK

#[async_trait]
impl ProcessorTrait for EventsProcessor {
    async fn process_transactions(
        ...
    ) -> anyhow::Result<ProcessingResult> {
        // Extract events from transactions
        let events: Vec<EventModel> = process_events(transactions);

        // Store the events in the database
        let tx_result = insert_to_db(
            self.get_pool(),
            self.name(),
            start_version,
            end_version,
            &events,
            &self.per_table_chunk_sizes,
        )
        .await;

        return tx_result;
    }
}

async fn insert_to_db(
    conn: ArcDbPool,
    name: &'static str,
    start_version: u64,
    end_version: u64,
    events: &[EventModel],
    per_table_chunk_sizes: &AHashMap<String, usize>,
) -> Result<(), diesel::result::Error> {
    tracing::trace!(
        name = name,
        start_version = start_version,
        end_version = end_version,
        "Inserting to db",
    );
    execute_in_chunks(
        conn,
        insert_events_query,
        events,
        get_config_table_chunk_size::<EventModel>("events", per_table_chunk_sizes),
    )
    .await?;
    Ok(())
}

// TODO: Update the step name
pub struct EventsExtractor
where
    Self: Sized + Send + 'static, {}

#[async_trait]
impl Processable for EventsExtractor {
    type Input = Vec<Transaction>;
    // TODO: Update the output type
    // This should be the data model you're extracting from the transactions
    type Output = Vec<EventModel>;
    type RunType = AsyncRunType;

    async fn process(
        &mut self,
        item: TransactionContext<Vec<Transaction>>,
    ) -> Result<Option<TransactionContext<Vec<EventModel>>>, ProcessorError> {
        // TODO: Update extraction logic.
        // This should be the same as the extraction logic in the old `process_transactions` method
        let events = item
            .data
            .par_iter()
            .map(|txn| {
                process_events(txn)
            })
            .flatten()
            .collect::<Vec<EventModel>>();

        Ok(Some(TransactionContext {
            data: events,
            metadata: item.metadata,
        }))
    }
}

pub struct EventsStorer
where
    Self: Sized + Send + 'static,
{
    conn_pool: ArcDbPool,
    processor_config: DefaultProcessorConfig,
}

impl EventsStorer {
    pub fn new(conn_pool: ArcDbPool, processor_config: DefaultProcessorConfig) -> Self {
        Self {
            conn_pool,
            processor_config,
        }
    }
}

#[async_trait]
// TODO: Update step name
impl Processable for EventsStorer {
    // TODO: Update input type for the step.
    // The input type should match the output type of the extractor step.
    type Input = Vec<EventModel>;
    type Output = ();
    type RunType = AsyncRunType;

    async fn process(
        &mut self,
        events: TransactionContext<Vec<EventModel>>,
    ) -> Result<Option<TransactionContext<()>>, ProcessorError> {
        let per_table_chunk_sizes: AHashMap<String, usize> = AHashMap::new();
        let execute_res = execute_in_chunks(
            self.conn_pool.clone(),
            // TODO: Update this to the insertion query of your old processor
            insert_events_query,
            &events.data,
            get_config_table_chunk_size::<EventModel>("events", &per_table_chunk_sizes),
        )
        .await;
        match execute_res {
            Ok(_) => {
                Ok(Some(TransactionContext {
                    data: (),
                    metadata: events.metadata,
                }))
            },
            Err(e) => Err(ProcessorError::DBStoreError {
                message: format!(
                    "Failed to store events versions {} to {}: {:?}",
                    events.metadata.start_version, events.metadata.end_version, e,
                ),
                query: None,
            }),
        }
    }
}

impl AsyncStep for EventsStorer {}

impl NamedStep for EventsStorer {
    fn name(&self) -> String {
        "EventsStorer".to_string()
    }
}

// TODO: Update processor name
pub struct EventsProcessor {
    pub config: IndexerProcessorConfig,
    pub db_pool: ArcDbPool,
    // If you have any other fields in your processor, add them here
    // You can instantiate them accordingly in the processor's `new` method
}

pub async fn run_processor(self) -> Result<()> {
    {...}

    // Define processor steps
    let transaction_stream_config = self.config.transaction_stream_config.clone();
    let transaction_stream = TransactionStreamStep::new(TransactionStreamConfig {
        starting_version: Some(starting_version),
        ..transaction_stream_config
    })
    .await?;
    // TODO: Replace the next 2 lines with your steps
    let events_extractor = EventsExtractor {};
    let events_storer = EventsStorer::new(self.db_pool.clone());

    let version_tracker = VersionTrackerStep::new(
        get_processor_status_saver(self.db_pool.clone(), self.config.clone()),
        DEFAULT_UPDATE_PROCESSOR_STATUS_SECS,
    );

    // Connect processor steps together
    let (_, buffer_receiver) = ProcessorBuilder::new_with_inputless_first_step(
        transaction_stream.into_runnable_step(),
    )
    // TODO: Replace the next 2 lines with your steps
    .connect_to(events_extractor.into_runnable_step(), 10)
    .connect_to(events_storer.into_runnable_step(), 10)
    .connect_to(version_tracker.into_runnable_step(), 10)
    .end_and_return_output_receiver(10);

    {...}
}

health_check_port: 8085
server_config:
  processor_config:
    # TODO: Update with processor type
    type: "events_processor"
  transaction_stream_config:
    indexer_grpc_data_service_address: "https://grpc.testnet.aptoslabs.com:443"
    # TODO: Update auth token
    auth_token: "AUTH_TOKEN"
    # TODO: Update with processor name
    request_name_header: "events-processor"
  db_config:
    # TODO: Update with your database connection string
    postgres_connection_string: postgresql://postgres:@localhost:5432/example
  # backfill_config:
  #   backfill_alias: "events_processor_backfill_1"

health_check_port: 8085
server_config:
  processor_config:
    # TODO: Update with processor type
    type: "events_processor"
  transaction_stream_config:
    indexer_grpc_data_service_address: "https://grpc.testnet.aptoslabs.com:443"
    # TODO: Update with backfill version
    starting_version: {backfill version}
    # TODO: Update auth token
    auth_token: "AUTH_TOKEN"
    # TODO: Update with processor name
    request_name_header: "events-processor"
  db_config:
    # TODO: Update with your database connection string
    postgres_connection_string: postgresql://postgres:@localhost:5432/example
  backfill_config:
    # TODO: Update with your backfill alias. This should be unique for each backfill
    backfill_alias: "events_processor_backfill_1"