Migrate auctions to new database table

crates/autopilot/src/arguments.rs

@@ -233,6 +233,11 @@ pub struct Arguments {
     /// Archive node URL used to index CoW AMM
     #[clap(long, env)]
     pub archive_node_url: Option<Url>,
+
+    /// Whether the migration from solver_competition table to new competition
+    /// tables should be run
+    #[clap(long, env, action = clap::ArgAction::Set, default_value = "false")]
+    pub migrate_auctions: bool,
 }
 
 impl std::fmt::Display for Arguments {
@@ -278,6 +283,7 @@ impl std::fmt::Display for Arguments {
             run_loop_native_price_timeout,
             max_winners_per_auction,
             archive_node_url,
+            migrate_auctions,
         } = self;
 
         write!(f, "{}", shared)?;
@@ -357,6 +363,7 @@ impl std::fmt::Display for Arguments {
         )?;
         writeln!(f, "max_winners_per_auction: {:?}", max_winners_per_auction)?;
         writeln!(f, "archive_node_url: {:?}", archive_node_url)?;
+        writeln!(f, "migrate_auctions: {:?}", migrate_auctions)?;
         Ok(())
     }
 }

crates/autopilot/src/infra/persistence/mod.rs

@@ -802,6 +802,92 @@ impl Persistence {
         ex.commit().await?;
         Ok(())
     }
+
+    pub async fn populate_historic_auctions(&self) -> Result<(), DatabaseError> {
+        const BATCH_SIZE: i64 = 50;
+
+        let mut ex = self.postgres.pool.begin().await?;
+
+        // find entry in `competition_auctions` with the lowest auction_id, as a
+        // starting point
+        let current_auction_id: Option<i64> =
+            sqlx::query_scalar::<_, Option<i64>>("SELECT MIN(id) FROM competition_auctions;")
+                .fetch_one(ex.deref_mut())
+                .await
+                .context("fetch lowest auction id")?;
+
+        let Some(mut current_auction_id) = current_auction_id else {
+            tracing::info!("competition_auctions is empty, nothing to process");
+            return Ok(());
+        };
+
+        loop {
+            tracing::debug!(
+                auction_id = current_auction_id,
+                "populating historic auctions from auction"
+            );
+
+            // fetch the next batch of auctions
+            let competitions: Vec<database::solver_competition::RichSolverCompetition> =
+                database::solver_competition::fetch_batch(&mut ex, current_auction_id, BATCH_SIZE)
+                    .await?;
+
+            if competitions.is_empty() {
+                tracing::info!("no more auctions to process");
+                break;
+            }
+
+            tracing::debug!(competitions = ?competitions.len(), "competitions fetched");
+
+            for solver_competition in &competitions {
+                let competition: model::solver_competition::SolverCompetitionDB =
+                    serde_json::from_value(solver_competition.json.clone())
+                        .context("deserialize SolverCompetitionDB")?;
+
+                // populate historic auctions
+                let auction = database::auction::Auction {
+                    id: solver_competition.id,
+                    block: i64::try_from(competition.auction_start_block)
+                        .context("block overflow")?,
+                    deadline: solver_competition.deadline,
+                    order_uids: competition
+                        .auction
+                        .orders
+                        .iter()
+                        .map(|order| ByteArray(order.0))
+                        .collect(),
+                    price_tokens: competition
+                        .auction
+                        .prices
+                        .keys()
+                        .map(|token| ByteArray(token.0))
+                        .collect(),
+                    price_values: competition
+                        .auction
+                        .prices
+                        .values()
+                        .map(u256_to_big_decimal)
+                        .collect(),
+                    surplus_capturing_jit_order_owners: solver_competition
+                        .surplus_capturing_jit_order_owners
+                        .clone(),
+                };
+
+                if let Err(err) = database::auction::save(&mut ex, auction).await {
+                    tracing::warn!(?err, auction_id = ?solver_competition.id, "failed to save auction");
+                }
+            }
+
+            // commit each batch separately
+            ex.commit().await?;
+            ex = self.postgres.pool.begin().await?;
+
+            // update the current auction id
+            current_auction_id = competitions.last().unwrap().id;
+        }
+
+        Ok(())
+    }
 }
 
 #[derive(prometheus_metric_storage::MetricStorage)]

crates/autopilot/src/run.rs

@@ -449,6 +449,11 @@ pub async fn run(args: Arguments) {
             .instrument(tracing::info_span!("order_events_cleaner")),
     );
 
+    if args.migrate_auctions {
+        let persistence_clone = persistence.clone();
+        tokio::spawn(async move { persistence_clone.populate_historic_auctions().await });
+    }
+
     let market_makable_token_list_configuration = TokenListConfiguration {
         url: args.trusted_tokens_url,
         update_interval: args.trusted_tokens_update_interval,

crates/database/src/solver_competition.rs

@@ -298,6 +298,43 @@ pub async fn fetch(
     Ok(solutions)
 }
 
+#[derive(Clone, Debug, sqlx::FromRow)]
+pub struct RichSolverCompetition {
+    pub id: AuctionId,
+    pub json: JsonValue,
+    pub deadline: i64,
+    pub surplus_capturing_jit_order_owners: Vec<crate::Address>,
+}
+
+/// Migrate all the auctions from the solver_competitions table to the auctions
+/// table. This is a one-time migration.
+///
+/// Entries are fetched going from higher auction_id to lower auction_id.
+pub async fn fetch_batch(
+    ex: &mut PgConnection,
+    auction_id: AuctionId,
+    batch_size: i64,
+) -> Result<Vec<RichSolverCompetition>, sqlx::Error> {
+    const QUERY: &str = r#"
+        SELECT 
+        sc.id as id, 
+        sc.json as json, 
+        COALESCE(ss.block_deadline, 0) AS deadline,
+        COALESCE(jit.owners, ARRAY[]::bytea[]) AS surplus_capturing_jit_order_owners
+        FROM solver_competitions sc
+        LEFT JOIN settlement_scores ss ON sc.id = ss.auction_id
+        LEFT JOIN surplus_capturing_jit_order_owners jit ON sc.id = jit.auction_id
+        WHERE sc.id < $1
+        ORDER BY sc.id DESC
+        LIMIT $2;"#;
+
+    sqlx::query_as(QUERY)
+        .bind(auction_id)
+        .bind(batch_size)
+        .fetch_all(ex)
+        .await
+}
+
 #[cfg(test)]
 mod tests {
     use {