use crate::pool::{
    Connection, ConnectionManager, JobEnqueueResult, ManagedConnection, Transaction,
};
use crate::selector::{CompileTestCase, RuntimeTestCase};
use crate::{
    ArtifactId, Benchmark, BenchmarkJob, BenchmarkJobConclusion, BenchmarkRequest,
    BenchmarkRequestIndex, BenchmarkRequestInsertResult, BenchmarkRequestStatus,
    BenchmarkRequestWithErrors, CodegenBackend, CollectionId, CollectorConfig, Commit, CommitType,
    CompileBenchmark, Date, PendingBenchmarkRequests, Profile, QueueRuntimeConfig, Repo, Tag,
    Target,
};
use crate::{ArtifactIdNumber, Index};
use chrono::{TimeZone, Utc};
use hashbrown::{HashMap, HashSet};
use rusqlite::params;
use rusqlite::OptionalExtension;
use std::path::PathBuf;
use std::str::FromStr;
use std::sync::Mutex;
use std::sync::Once;
use std::time::Duration;

pub struct SqliteTransaction<'a> {
    conn: &'a mut SqliteConnection,
    finished: bool,
}

#[async_trait::async_trait]
impl Transaction for SqliteTransaction<'_> {
    async fn commit(mut self: Box<Self>) -> Result<(), anyhow::Error> {
        self.finished = true;
        Ok(self.conn.raw().execute_batch("COMMIT")?)
    }

    async fn finish(mut self: Box<Self>) -> Result<(), anyhow::Error> {
        self.finished = true;
        Ok(self.conn.raw().execute_batch("ROLLBACK")?)
    }
    fn conn(&mut self) -> &mut dyn Connection {
        &mut *self.conn
    }
    fn conn_ref(&self) -> &dyn Connection {
        &*self.conn
    }
}

impl std::ops::Deref for SqliteTransaction<'_> {
    type Target = dyn Connection;
    fn deref(&self) -> &Self::Target {
        &*self.conn
    }
}

impl std::ops::DerefMut for SqliteTransaction<'_> {
    fn deref_mut(&mut self) -> &mut Self::Target {
        self.conn
    }
}

impl Drop for SqliteTransaction<'_> {
    fn drop(&mut self) {
        if !self.finished {
            self.conn.raw().execute_batch("ROLLBACK").unwrap();
        }
    }
}

pub struct Sqlite(PathBuf, Once);

impl Sqlite {
    pub fn new(path: PathBuf) -> Self {
        Sqlite(path, Once::new())
    }
}

struct Migration {
    /// One or more SQL statements, each terminated by a semicolon.
    sql: &'static str,

    /// If false, indicates that foreign key checking should be delayed until after execution of
    /// the migration SQL, and foreign key `ON UPDATE` and `ON DELETE` actions disabled completely.
    foreign_key_constraints_enabled: bool,
}

impl Migration {
    /// Returns a `Migration` with foreign key constraints enabled during execution.
    const fn new(sql: &'static str) -> Migration {
        Migration {
            sql,
            foreign_key_constraints_enabled: true,
        }
    }

    /// Returns a `Migration` with foreign key checking delayed until after execution, and foreign
    /// key `ON UPDATE` and `ON DELETE` actions disabled completely.
    ///
    /// SQLite has limited `ALTER TABLE` capabilities, so some schema alterations require the
    /// approach of replacing a table with a new one having the desired schema. Because there might
    /// be other tables with foreign key constraints on the table, these constraints need to be
    /// disabled during execution of such migration SQL, and reenabled after. Otherwise, dropping
    /// the old table may trigger `ON DELETE` actions in the referencing tables. See [SQLite
    /// documentation](https://www.sqlite.org/lang_altertable.html) for more information.
    const fn without_foreign_key_constraints(sql: &'static str) -> Migration {
        Migration {
            sql,
            foreign_key_constraints_enabled: false,
        }
    }

    fn execute(&self, conn: &mut rusqlite::Connection, migration_id: i32) {
        if self.foreign_key_constraints_enabled {
            let tx = conn.transaction().unwrap();
            tx.execute_batch(self.sql).unwrap();
            tx.pragma_update(None, "user_version", migration_id)
                .unwrap();
            tx.commit().unwrap();
            return;
        }

        // The following steps are reproduced from https://www.sqlite.org/lang_altertable.html,
        // from the section titled, "Making Other Kinds Of Table Schema Changes".

        // 1.  If foreign key constraints are enabled, disable them using PRAGMA foreign_keys=OFF.
        conn.pragma_update(None, "foreign_keys", "OFF").unwrap();

        // 2.  Start a transaction.
        let tx = conn.transaction().unwrap();

        // The migration SQL is responsible for steps 3 through 9.

        // 3.  Remember the format of all indexes, triggers, and views associated with table X.
        //     This information will be needed in step 8 below. One way to do this is to run a
        //     query like the following: SELECT type, sql FROM sqlite_schema WHERE tbl_name='X'.
        //
        // 4.  Use CREATE TABLE to construct a new table "new_X" that is in the desired revised
        //     format of table X. Make sure that the name "new_X" does not collide with any
        //     existing table name, of course.
        //
        // 5.  Transfer content from X into new_X using a statement like: INSERT INTO new_X SELECT
        //     ... FROM X.
        //
        // 6.  Drop the old table X: DROP TABLE X.
        //
        // 7.  Change the name of new_X to X using: ALTER TABLE new_X RENAME TO X.
        //
        // 8.  Use CREATE INDEX, CREATE TRIGGER, and CREATE VIEW to reconstruct indexes, triggers,
        //     and views associated with table X. Perhaps use the old format of the triggers,
        //     indexes, and views saved from step 3 above as a guide, making changes as appropriate
        //     for the alteration.
        //
        // 9.  If any views refer to table X in a way that is affected by the schema change, then
        //     drop those views using DROP VIEW and recreate them with whatever changes are
        //     necessary to accommodate the schema change using CREATE VIEW.
        tx.execute_batch(self.sql).unwrap();

        // 10. If foreign key constraints were originally enabled then run PRAGMA foreign_key_check
        //     to verify that the schema change did not break any foreign key constraints.
        tx.pragma_query(None, "foreign_key_check", |row| {
            let table: String = row.get_unwrap(0);
            let row_id: Option<i64> = row.get_unwrap(1);
            let foreign_table: String = row.get_unwrap(2);
            let fk_idx: i64 = row.get_unwrap(3);

            tx.query_row::<(), _, _>(
                "select * from pragma_foreign_key_list(?) where id = ?",
                params![&table, &fk_idx],
                |row| {
                    let col: String = row.get_unwrap(3);
                    let foreign_col: String = row.get_unwrap(4);
                    panic!(
                        "Foreign key violation encountered during migration\n\
                            table: {table},\n\
                            column: {col},\n\
                            row_id: {row_id:?},\n\
                            foreign table: {foreign_table},\n\
                            foreign column: {foreign_col}\n\
                            migration ID: {migration_id}\n",
                    );
                },
            )
            .unwrap();
            Ok(())
        })
        .unwrap();

        tx.pragma_update(None, "user_version", migration_id)
            .unwrap();

        // 11. Commit the transaction started in step 2.
        tx.commit().unwrap();

        // 12. If foreign keys constraints were originally enabled, reenable them now.
        conn.pragma_update(None, "foreign_keys", "ON").unwrap();
    }
}

static MIGRATIONS: &[Migration] = &[
    Migration::new(""),
    Migration::new(
        r#"
        create table benchmark(
            name text primary key,
            -- Whether this benchmark supports stable
            stabilized bool not null
        );
        create table artifact(
            id integer primary key not null,
            name text not null unique,
            date integer,
            type text not null
        );
        create table collection(
            id integer primary key not null
        );
        create table error_series(
            id integer primary key not null,
            crate text not null unique references benchmark(name) on delete cascade on update cascade
        );
        create table error(
            series integer not null references error_series(id) on delete cascade on update cascade,
            aid integer not null references artifact(id) on delete cascade on update cascade,
            error text,
            PRIMARY KEY(series, aid)
        );
        create table pstat_series(
            id integer primary key not null,
            crate text not null references benchmark(name) on delete cascade on update cascade,
            profile text not null,
            cache text not null,
            statistic text not null,
            UNIQUE(crate, profile, cache, statistic)
        );
        create table pstat(
            series integer references pstat_series(id) on delete cascade on update cascade,
            aid integer references artifact(id) on delete cascade on update cascade,
            cid integer references collection(id) on delete cascade on update cascade,
            value double not null,
            PRIMARY KEY(series, aid, cid)
        );
        create table self_profile_query_series(
            id integer primary key not null,
            crate text not null references benchmark(name) on delete cascade on update cascade,
            profile text not null,
            cache text not null,
            query text not null,
            UNIQUE(crate, profile, cache, query)
        );
        create table self_profile_query(
            series integer references self_profile_query_series(id) on delete cascade on update cascade,
            aid integer references artifact(id) on delete cascade on update cascade,
            cid integer references collection(id) on delete cascade on update cascade,
            self_time integer,
            blocked_time integer,
            incremental_load_time integer,
            number_of_cache_hits integer,
            invocation_count integer,
            PRIMARY KEY(series, aid, cid)
        );
        create table pull_request_builds(
            bors_sha text unique,
            pr integer not null,
            parent_sha text,
            complete boolean,
            requested timestamp without time zone
        );
        "#,
    ),
    Migration::new(
        r#"
        create table artifact_collection_duration(
            aid integer primary key not null references artifact(id) on delete cascade on update cascade,
            date_recorded timestamp without time zone not null,
            duration integer not null
        );
        "#,
    ),
    Migration::new(
        r#"
        create table collector_progress(
            aid integer not null references artifact(id) on delete cascade on update cascade,
            step text not null,
            start integer,
            end integer,
            UNIQUE(aid, step)
        );
        "#,
    ),
    Migration::new("alter table collection add column perf_commit text"),
    Migration::new("alter table pull_request_builds add column include text"),
    Migration::new("alter table pull_request_builds add column exclude text"),
    Migration::new("alter table pull_request_builds add column runs integer"),
    Migration::new(
        r#"
        create table rustc_compilation(
            aid integer references artifact(id) on delete cascade on update cascade,
            cid integer references collection(id) on delete cascade on update cascade,
            crate text not null,
            duration integer not null,
            PRIMARY KEY(aid, cid, crate)
        );
        "#,
    ),
    Migration::new("alter table pull_request_builds rename to pull_request_build"),
    Migration::new(
        r#"
        create table raw_self_profile(
            aid integer references artifact(id) on delete cascade on update cascade,
            cid integer references collection(id) on delete cascade on update cascade,
            crate text not null references benchmark(name) on delete cascade on update cascade,
            profile text not null,
            cache text not null,
            PRIMARY KEY(aid, cid, crate, profile, cache)
        );
        "#,
    ),
    // Add not null constraint to benchmark name.
    Migration::without_foreign_key_constraints(
        r#"
        create table benchmark_new(
            name text primary key not null,
            stabilized bool not null
        );
        insert into benchmark_new select * from benchmark where name is not null;
        drop table benchmark;
        alter table benchmark_new rename to benchmark;
        "#,
    ),
    Migration::new("alter table benchmark add column category text not null default ''"),
    Migration::new("alter table pull_request_build add column commit_date timestamp"),
    Migration::new(
        r#"
        create table runtime_pstat_series(
            id integer primary key not null,
            benchmark text not null,
            metric text not null,
            UNIQUE(benchmark, metric)
        );
        create table runtime_pstat(
            series integer references runtime_pstat_series(id) on delete cascade on update cascade,
            aid integer references artifact(id) on delete cascade on update cascade,
            cid integer references collection(id) on delete cascade on update cascade,
            value double not null,
            PRIMARY KEY(series, aid, cid)
        );
        "#,
    ),
    Migration::new(
        r#"
        create table error_new(
            aid integer references artifact(id) on delete cascade on update cascade,
            benchmark text not null,
            error text not null,
            primary key(aid, benchmark)
        );
        insert into error_new(aid, benchmark, error)
        select aid, crate, error
        from error
        join error_series es on error.series = es.id;

        drop table error;
        drop table error_series;
        alter table error_new rename to error;
    "#,
    ),
    Migration::new(
        r#"
        create table artifact_size(
            aid integer references artifact(id) on delete cascade on update cascade,
            component text not null,
            size integer not null,
            UNIQUE(aid, component)
        );
    "#,
    ),
    // Add codegen backend column and add it to the unique constraint.
    // Also rename cache to scenario and statistic to metric, while we're at it.
    Migration::without_foreign_key_constraints(
        r#"
        create table pstat_series_new(
            id integer primary key not null,
            crate text not null references benchmark(name) on delete cascade on update cascade,
            profile text not null,
            scenario text not null,
            backend text not null,
            metric text not null,
            UNIQUE(crate, profile, scenario, backend, metric)
        );
        insert into pstat_series_new select id, crate, profile, cache, 'llvm', statistic from pstat_series;
        drop table pstat_series;
        alter table pstat_series_new rename to pstat_series;
    "#,
    ),
    Migration::new("alter table pull_request_build add column backends text"),
    // Add target as a unique constraint, defaulting to 'x86_64-unknown-linux-gnu'
    Migration::without_foreign_key_constraints(
        r#"
        create table pstat_series_with_target(
            id integer primary key not null,
            crate text not null references benchmark(name) on delete cascade on update cascade,
            profile text not null,
            scenario text not null,
            backend text not null,
            target text not null default 'x86_64-unknown-linux-gnu',
            metric text not null,
            UNIQUE(crate, profile, scenario, backend, target, metric)
        );
        insert into pstat_series_with_target select id, crate, profile, scenario, backend, 'x86_64-unknown-linux-gnu', metric from pstat_series;
        drop table pstat_series;
        alter table pstat_series_with_target rename to pstat_series;
    "#,
    ),
    Migration::new(
        r#"
        CREATE TABLE error_new (
            id      INTEGER PRIMARY KEY,
            aid     INTEGER NOT NULL REFERENCES artifact(id) ON DELETE CASCADE ON UPDATE CASCADE,
            message TEXT NOT NULL,
            context TEXT NOT NULL,
            job_id  INTEGER
        );

        INSERT INTO
            error_new (aid, message, context)
        SELECT
            aid,
            error,
            benchmark
        FROM
            error;

        DROP TABLE error;
        ALTER TABLE error_new RENAME TO error;

        CREATE INDEX error_artifact_idx ON error(aid);
        "#,
    ),
    // Add runtime target as a unique constraint, defaulting to 'x86_64-unknown-linux-gnu'
    Migration::without_foreign_key_constraints(
        r#"
        CREATE TABLE runtime_pstat_series_with_target(
            id integer primary key not null,
            benchmark text not null,
            target text not null default 'x86_64-unknown-linux-gnu',
            metric text not null,
            UNIQUE(benchmark, target, metric)
        );
        INSERT INTO runtime_pstat_series_with_target SELECT id, benchmark, 'x86_64-unknown-linux-gnu', metric FROM runtime_pstat_series;
        DROP TABLE runtime_pstat_series;
        ALTER TABLE runtime_pstat_series_with_target RENAME TO runtime_pstat_series;
    "#,
    ),
    // Add json_value column to runtime_pstat for storing structured JSON metrics
    Migration::new("ALTER TABLE runtime_pstat ADD COLUMN json_value TEXT DEFAULT NULL"),
    // Add tag column to pstat_series and runtime_pstat_series
    Migration::without_foreign_key_constraints(
        r#"
        CREATE TABLE pstat_series_with_tag(
            id integer primary key not null,
            crate text not null references benchmark(name) on delete cascade on update cascade,
            profile text not null,
            scenario text not null,
            backend text not null,
            target text not null default 'x86_64-unknown-linux-gnu',
            tag text not null default 'default',
            metric text not null,
            UNIQUE(crate, profile, scenario, backend, target, tag, metric)
        );
        INSERT INTO pstat_series_with_tag SELECT id, crate, profile, scenario, backend, target, 'default', metric FROM pstat_series;
        DROP TABLE pstat_series;
        ALTER TABLE pstat_series_with_tag RENAME TO pstat_series;
    "#,
    ),
    Migration::without_foreign_key_constraints(
        r#"
        CREATE TABLE runtime_pstat_series_with_tag(
            id integer primary key not null,
            benchmark text not null,
            target text not null default 'x86_64-unknown-linux-gnu',
            tag text not null default 'default',
            metric text not null,
            UNIQUE(benchmark, target, tag, metric)
        );
        INSERT INTO runtime_pstat_series_with_tag SELECT id, benchmark, target, 'default', metric FROM runtime_pstat_series;
        DROP TABLE runtime_pstat_series;
        ALTER TABLE runtime_pstat_series_with_tag RENAME TO runtime_pstat_series;
    "#,
    ),
    // Move tag from pstat_series/runtime_pstat_series to the artifact table.
    Migration::without_foreign_key_constraints(
        r#"
        CREATE TABLE artifact_with_tag(
            id integer primary key not null,
            name text not null,
            date integer,
            type text not null,
            tag text not null default 'default',
            UNIQUE(name, tag)
        );
        INSERT INTO artifact_with_tag SELECT id, name, date, type, 'default' FROM artifact;
        DROP TABLE artifact;
        ALTER TABLE artifact_with_tag RENAME TO artifact;
    "#,
    ),
    Migration::without_foreign_key_constraints(
        r#"
        CREATE TABLE pstat_series_no_tag(
            id integer primary key not null,
            crate text not null references benchmark(name) on delete cascade on update cascade,
            profile text not null,
            scenario text not null,
            backend text not null,
            target text not null default 'x86_64-unknown-linux-gnu',
            metric text not null,
            UNIQUE(crate, profile, scenario, backend, target, metric)
        );
        INSERT INTO pstat_series_no_tag SELECT id, crate, profile, scenario, backend, target, metric FROM pstat_series;
        DROP TABLE pstat_series;
        ALTER TABLE pstat_series_no_tag RENAME TO pstat_series;
    "#,
    ),
    Migration::without_foreign_key_constraints(
        r#"
        CREATE TABLE runtime_pstat_series_no_tag(
            id integer primary key not null,
            benchmark text not null,
            target text not null default 'x86_64-unknown-linux-gnu',
            metric text not null,
            UNIQUE(benchmark, target, metric)
        );
        INSERT INTO runtime_pstat_series_no_tag SELECT id, benchmark, target, metric FROM runtime_pstat_series;
        DROP TABLE runtime_pstat_series;
        ALTER TABLE runtime_pstat_series_no_tag RENAME TO runtime_pstat_series;
    "#,
    ),
    Migration::without_foreign_key_constraints(
        r#"
        CREATE TABLE artifact_with_repo(
            id integer primary key not null,
            name text not null,
            date integer,
            type text not null,
            tag text not null default 'default',
            repo text not null default 'rust',
            UNIQUE(name, tag, repo)
        );
        INSERT INTO artifact_with_repo SELECT id, name, date, type, tag, 'rust' FROM artifact;
        DROP TABLE artifact;
        ALTER TABLE artifact_with_repo RENAME TO artifact;
    "#,
    ),
];

#[async_trait::async_trait]
impl ConnectionManager for Sqlite {
    type Connection = Mutex<rusqlite::Connection>;
    async fn open(&self) -> Self::Connection {
        let mut conn = rusqlite::Connection::open(&self.0).unwrap();
        conn.pragma_update(None, "cache_size", -128000).unwrap();
        conn.pragma_update(None, "journal_mode", "WAL").unwrap();
        conn.pragma_update(None, "foreign_keys", "ON").unwrap();

        self.1.call_once(|| {
            let version: i32 = conn
                .query_row(
                    "select user_version from pragma_user_version;",
                    params![],
                    |row| row.get(0),
                )
                .unwrap();
            for (mid, migration) in MIGRATIONS.iter().enumerate().skip(version as usize + 1) {
                migration.execute(&mut conn, mid as i32);
            }
        });

        Mutex::new(conn)
    }
    async fn is_valid(&self, conn: &mut Self::Connection) -> bool {
        conn.get_mut()
            .unwrap_or_else(|e| e.into_inner())
            .execute_batch("")
            .is_ok()
    }
}

pub struct SqliteConnection {
    conn: ManagedConnection<Mutex<rusqlite::Connection>>,
}

fn assert_sync<T: Sync>() {}

impl SqliteConnection {
    pub fn new(conn: ManagedConnection<Mutex<rusqlite::Connection>>) -> Self {
        assert_sync::<Self>();
        Self { conn }
    }

    pub fn raw(&mut self) -> &mut rusqlite::Connection {
        self.conn.get_mut().unwrap_or_else(|e| e.into_inner())
    }
    pub fn raw_ref(&self) -> std::sync::MutexGuard<'_, rusqlite::Connection> {
        self.conn.lock().unwrap_or_else(|e| e.into_inner())
    }
}

macro_rules! no_queue_implementation_abort {
    () => {
        panic!(
            "Queueing for SQLite has not been implemented; if you want to test the queueing \
             functionality please use Postgres. Presuming you have Docker installed, at the \
             root of the repo you can run `make start-postgres` to spin up a Postgres \
             database."
        )
    };
}

/// Pre-computed maps and placeholders for batch IN-clause queries on runtime_pstat.
struct BatchQueryParams {
    sid_map: HashMap<u32, usize>,
    aid_map: HashMap<u32, usize>,
    sid_placeholders: String,
    aid_placeholders: String,
    valid_aids: Vec<u32>,
}

impl BatchQueryParams {
    /// Returns `None` if there are no valid series or artifact IDs (caller should return early).
    fn new(series_ids: &[u32], artifact_ids: &[Option<ArtifactIdNumber>]) -> Option<Self> {
        let valid_aids: Vec<u32> = artifact_ids.iter().filter_map(|a| a.map(|a| a.0)).collect();

        if series_ids.is_empty() || valid_aids.is_empty() {
            return None;
        }

        let sid_map: HashMap<u32, usize> = series_ids
            .iter()
            .enumerate()
            .map(|(i, &sid)| (sid, i))
            .collect();
        let aid_map: HashMap<u32, usize> = artifact_ids
            .iter()
            .enumerate()
            .filter_map(|(i, a)| a.map(|a| (a.0, i)))
            .collect();

        let sid_placeholders: String = series_ids.iter().map(|_| "?").collect::<Vec<_>>().join(",");
        let aid_placeholders: String = valid_aids.iter().map(|_| "?").collect::<Vec<_>>().join(",");

        Some(Self {
            sid_map,
            aid_map,
            sid_placeholders,
            aid_placeholders,
            valid_aids,
        })
    }

    fn bind_params(&self, stmt: &mut rusqlite::CachedStatement, series_ids: &[u32]) {
        let mut idx = 1;
        for sid in series_ids {
            stmt.raw_bind_parameter(idx, *sid as i64).unwrap();
            idx += 1;
        }
        for aid in &self.valid_aids {
            stmt.raw_bind_parameter(idx, *aid as i64).unwrap();
            idx += 1;
        }
    }
}

#[async_trait::async_trait]
impl Connection for SqliteConnection {
    async fn maybe_create_indices(&mut self) {
        self.raw()
            .execute_batch(
                "
            create index if not exists pstat_on_series_aid on pstat(series, aid);
        ",
            )
            .unwrap();
    }

    async fn transaction(&mut self) -> Box<dyn Transaction + '_> {
        self.raw().execute_batch("BEGIN DEFERRED").unwrap();
        Box::new(SqliteTransaction {
            conn: self,
            finished: false,
        })
    }

    async fn load_index(&mut self) -> Index {
        let commits = self
            .raw()
            .prepare(
                "select id, name, date, type, tag, repo from artifact where type = 'master' or type = 'try'",
            )
            .unwrap()
            .query_map(params![], |row| {
                Ok((
                    row.get::<_, i32>(0)? as u32,
                    (
                        Repo::from(row.get::<_, String>(5)?.as_str()),
                        Commit {
                            sha: row.get::<_, String>(1)?.as_str().into(),
                            date: {
                                let timestamp: Option<i64> = row.get(2)?;
                                match timestamp {
                                    Some(t) => Date(Utc.timestamp_opt(t, 0).unwrap()),
                                    None => Date(Utc.with_ymd_and_hms(2001, 1, 1, 0, 0, 0).unwrap()),
                                }
                            },
                            r#type: CommitType::from_str(&row.get::<_, String>(3)?).unwrap(),
                        },
                        row.get::<_, String>(4)?.as_str().parse::<Tag>().unwrap(),
                    ),
                ))
            })
            .unwrap()
            .map(|r| r.unwrap())
            .collect();
        let artifacts = self
            .raw()
            .prepare("select id, name, tag, repo from artifact where type = 'release'")
            .unwrap()
            .query_map(params![], |row| {
                Ok((
                    row.get::<_, i32>(0)? as u32,
                    (
                        Repo::from(row.get::<_, String>(3)?.as_str()),
                        row.get::<_, String>(1)?.into_boxed_str(),
                        row.get::<_, String>(2)?.as_str().parse::<Tag>().unwrap(),
                    ),
                ))
            })
            .unwrap()
            .map(|r| r.unwrap())
            .collect();
        let pstat_series = self
            .raw()
            .prepare(
                "select id, crate, profile, scenario, backend, target, metric from pstat_series;",
            )
            .unwrap()
            .query_map(params![], |row| {
                Ok((
                    row.get::<_, i32>(0)? as u32,
                    (
                        Benchmark::from(row.get::<_, String>(1)?.as_str()),
                        Profile::from_str(row.get::<_, String>(2)?.as_str()).unwrap(),
                        row.get::<_, String>(3)?.as_str().parse().unwrap(),
                        CodegenBackend::from_str(row.get::<_, String>(4)?.as_str()).unwrap(),
                        Target::from_str(row.get::<_, String>(5)?.as_str()).unwrap(),
                        row.get::<_, String>(6)?.as_str().into(),
                    ),
                ))
            })
            .unwrap()
            .map(|r| r.unwrap())
            .collect();
        let runtime_pstat_series = self
            .raw()
            .prepare("SELECT id, benchmark, target, metric FROM runtime_pstat_series;")
            .unwrap()
            .query_map(params![], |row| {
                Ok((
                    row.get::<_, i32>(0)? as u32,
                    (
                        row.get::<_, String>(1)?.as_str().into(),
                        Target::from_str(row.get::<_, String>(2)?.as_str()).unwrap(),
                        row.get::<_, String>(3)?.as_str().into(),
                    ),
                ))
            })
            .unwrap()
            .map(|r| r.unwrap())
            .collect();
        Index {
            commits,
            artifacts,
            pstat_series,
            runtime_pstat_series,
        }
    }

    async fn record_compile_benchmark(
        &self,
        benchmark: &str,
        supports_stable: Option<bool>,
        category: String,
    ) {
        if let Some(stable) = supports_stable {
            self.raw_ref()
                .execute(
                    "insert into benchmark (name, stabilized, category) VALUES (?, ?, ?)
                ON CONFLICT (name) do update set stabilized = excluded.stabilized, category = excluded.category",
                    params![benchmark, stable, category],
                )
                .unwrap();
        } else {
            self.raw_ref()
                .execute(
                    "insert into benchmark (name, stabilized, category) VALUES (?, ?, ?)
                ON CONFLICT (name) do update set category = excluded.category",
                    params![benchmark, false, category],
                )
                .unwrap();
        }
    }

    async fn get_compile_benchmarks(&self) -> Vec<CompileBenchmark> {
        let conn = self.raw_ref();
        let mut query = conn
            .prepare_cached("select name, category from benchmark")
            .unwrap();
        let rows = query
            .query_map([], |row| {
                Ok(CompileBenchmark {
                    name: row.get(0)?,
                    category: row.get::<_, String>(1)?,
                })
            })
            .unwrap();
        let mut benchmarks = Vec::new();
        for row in rows {
            benchmarks.push(row.unwrap());
        }
        benchmarks
    }

    async fn get_repos(&self) -> Vec<String> {
        let conn = self.raw_ref();
        let mut query = conn
            .prepare_cached("SELECT DISTINCT repo FROM artifact ORDER BY repo")
            .unwrap();
        let rows = query.query_map([], |row| row.get::<_, String>(0)).unwrap();
        let mut repos = Vec::new();
        for row in rows {
            repos.push(row.unwrap());
        }
        repos
    }

    async fn get_benchmarked_commits(
        &self,
        repo: Option<&str>,
    ) -> Vec<(String, String, Vec<String>)> {
        let conn = self.raw_ref();
        let (sql, params) = if let Some(repo) = repo {
            (
                "SELECT a.name, a.date, GROUP_CONCAT(DISTINCT a.tag) AS tags \
                 FROM artifact a \
                 WHERE a.type = 'master' AND a.repo = ? \
                 GROUP BY a.name, a.date \
                 ORDER BY a.date DESC",
                vec![Box::new(repo.to_string()) as Box<dyn rusqlite::types::ToSql>],
            )
        } else {
            (
                "SELECT a.name, a.date, GROUP_CONCAT(DISTINCT a.tag) AS tags \
                 FROM artifact a \
                 WHERE a.type = 'master' \
                 GROUP BY a.name, a.date \
                 ORDER BY a.date DESC",
                vec![],
            )
        };
        let params_refs: Vec<&dyn rusqlite::types::ToSql> = params
            .iter()
            .map(|p| p.as_ref() as &dyn rusqlite::types::ToSql)
            .collect();
        let mut query = conn.prepare_cached(sql).unwrap();
        let rows = query
            .query_map(params_refs.as_slice(), |row| {
                let sha: String = row.get(0)?;
                let date_val: Option<i64> = row.get(1)?;
                let date_str = date_val.map(|v| v.to_string());
                let tags_str: Option<String> = row.get(2)?;
                let tags = tags_str
                    .map(|s| s.split(',').map(String::from).collect())
                    .unwrap_or_default();
                Ok((sha, date_str.unwrap_or_default(), tags))
            })
            .unwrap();
        let mut commits = Vec::new();
        for row in rows {
            commits.push(row.unwrap());
        }
        commits
    }
    async fn artifact_by_name(&self, artifact: &str) -> Option<ArtifactId> {
        let (date, ty, repo) = self
            .raw_ref()
            .prepare("select date, type, repo from artifact where name = ? order by case when repo = 'rust' then 0 else 1 end, id limit 1")
            .unwrap()
            .query_row(params![&artifact], |r| {
                let date = r.get::<_, Option<i64>>(0)?;
                let ty = r.get::<_, String>(1)?;
                let repo = r.get::<_, String>(2)?;
                Ok((date, ty, repo))
            })
            .optional()
            .unwrap()?;

        Some(parse_artifact_id(
            ty.as_str(),
            artifact,
            date,
            repo.as_str(),
        ))
    }

    async fn collection_id(&self, version: &str) -> CollectionId {
        let raw = self.raw_ref();
        raw.execute(
            "insert into collection (perf_commit) values (?)",
            params![version],
        )
        .unwrap();
        CollectionId(
            raw.query_row(
                "select id from collection where rowid = last_insert_rowid()",
                params![],
                |r| r.get(0),
            )
            .unwrap(),
        )
    }
    async fn artifact_id(&self, artifact: &crate::ArtifactId, tag: &str) -> ArtifactIdNumber {
        let info = artifact.info();
        let repo = artifact.repo();

        self.raw_ref()
            .execute(
                "insert or ignore into artifact (name, date, type, tag, repo) VALUES (?, ?, ?, ?, ?)",
                params![
                    &info.name,
                    &info.date.map(|d| d.timestamp()),
                    &info.kind,
                    &tag,
                    &repo
                ],
            )
            .unwrap();
        ArtifactIdNumber(
            self.raw_ref()
                .query_row(
                    "select id from artifact where name = $1 and tag = $2 and repo = $3",
                    params![&info.name, &tag, &repo],
                    |r| r.get::<_, i32>(0),
                )
                .unwrap() as u32,
        )
    }
    async fn record_statistic(
        &self,
        collection: CollectionId,
        artifact: ArtifactIdNumber,
        benchmark: &str,
        profile: Profile,
        scenario: crate::Scenario,
        backend: CodegenBackend,
        target: Target,
        metric: &str,
        value: f64,
    ) {
        let profile = profile.to_string();
        let scenario = scenario.to_string();
        let backend = backend.to_string();
        let target = target.to_string();
        self.raw_ref().execute("insert or ignore into pstat_series (crate, profile, scenario, backend, target, metric) VALUES (?, ?, ?, ?, ?, ?)", params![
            &benchmark,
            &profile,
            &scenario,
            &backend,
            &target,
            &metric,
        ]).unwrap();
        let sid: i32 = self.raw_ref().query_row("select id from pstat_series where crate = ? and profile = ? and scenario = ? and backend = ? and target = ? and metric = ?", params![
            &benchmark,
            &profile,
            &scenario,
            &backend,
            &target,
            &metric,
        ], |r| r.get(0)).unwrap();
        self.raw_ref()
            .execute(
                "insert into pstat (series, aid, cid, value) VALUES (?, ?, ?, ?)",
                params![&sid, &artifact.0, &collection.0, &value],
            )
            .unwrap();
    }
    async fn record_runtime_statistic(
        &self,
        collection: CollectionId,
        artifact: ArtifactIdNumber,
        benchmark: &str,
        metric: &str,
        target: Target,
        value: f64,
    ) {
        let target = target.to_string();
        self.raw_ref()
            .execute(
                "INSERT OR IGNORE INTO runtime_pstat_series (benchmark, target, metric) VALUES (?, ?, ?)",
                params![&benchmark, &target, &metric],
            )
            .unwrap();
        let sid: i32 = self
            .raw_ref()
            .query_row(
                "SELECT id FROM runtime_pstat_series WHERE benchmark = ? AND target = ? AND metric = ?",
                params![&benchmark, &target, &metric],
                |r| r.get(0),
            )
            .unwrap();
        self.raw_ref()
            .execute(
                "INSERT INTO runtime_pstat (series, aid, cid, value) VALUES (?, ?, ?, ?)",
                params![&sid, &artifact.0, &collection.0, &value],
            )
            .unwrap();
    }

    async fn record_runtime_json_statistic(
        &self,
        collection: CollectionId,
        artifact: ArtifactIdNumber,
        benchmark: &str,
        metric: &str,
        target: Target,
        json_value: &str,
    ) {
        let target = target.to_string();
        self.raw_ref()
            .execute(
                "INSERT OR IGNORE INTO runtime_pstat_series (benchmark, target, metric) VALUES (?, ?, ?)",
                params![&benchmark, &target, &metric],
            )
            .unwrap();
        let sid: i32 = self
            .raw_ref()
            .query_row(
                "SELECT id FROM runtime_pstat_series WHERE benchmark = ? AND target = ? AND metric = ?",
                params![&benchmark, &target, &metric],
                |r| r.get(0),
            )
            .unwrap();
        self.raw_ref()
            .execute(
                "INSERT INTO runtime_pstat (series, aid, cid, value, json_value) VALUES (?, ?, ?, 0, ?)",
                params![&sid, &artifact.0, &collection.0, &json_value],
            )
            .unwrap();
    }

    async fn record_error(
        &self,
        artifact: ArtifactIdNumber,
        context: &str,
        message: &str,
        job_id: Option<u32>,
    ) {
        if job_id.is_some() {
            no_queue_implementation_abort!()
        }
        self.raw_ref()
            .execute(
                "insert into error (context, aid, message) VALUES (?, ?, ?)",
                params![context, &artifact.0, &message],
            )
            .unwrap();
    }
    async fn record_rustc_crate(
        &self,
        collection: CollectionId,
        artifact: ArtifactIdNumber,
        krate: &str,
        value: Duration,
    ) {
        self.raw_ref()
            .execute(
                "insert into rustc_compilation (aid, cid, crate, duration) VALUES (?, ?, ?, ?)",
                params![
                    &artifact.0,
                    &collection.0,
                    &krate,
                    &(value.as_nanos() as i64)
                ],
            )
            .unwrap();
    }

    async fn record_artifact_size(&self, artifact: ArtifactIdNumber, component: &str, size: u64) {
        self.raw_ref()
            .execute(
                "insert or replace into artifact_size (aid, component, size)\
                values (?, ?, ?)",
                params![&artifact.0, &component, &size],
            )
            .unwrap();
    }

    async fn get_artifact_size(&self, aid: ArtifactIdNumber) -> HashMap<String, u64> {
        self.raw_ref()
            .prepare("select component, size from artifact_size where aid = ?")
            .unwrap()
            .query_map(params![&aid.0], |row| {
                Ok((row.get::<_, String>(0)?, row.get::<_, u64>(1)?))
            })
            .unwrap()
            .map(|r| r.unwrap())
            .collect()
    }

    async fn get_bootstrap(&self, aids: &[ArtifactIdNumber]) -> Vec<Option<Duration>> {
        aids.iter()
            .map(|aid| {
                self.raw_ref()
                    .prepare(
                        "
                        select min(total)
                        from (
                            select sum(duration) as total
                            from rustc_compilation
                            where aid = ?
                            group by cid
                        )
                    ",
                    )
                    .unwrap()
                    .query_row(params![&aid.0], |row| {
                        Ok(Duration::from_nanos(row.get::<_, i64>(0)? as u64))
                    })
                    .optional()
                    .unwrap()
            })
            .collect()
    }

    async fn get_bootstrap_by_crate(
        &self,
        aids: &[ArtifactIdNumber],
    ) -> HashMap<String, Vec<Option<Duration>>> {
        let mut results = HashMap::new();

        for (idx, aid) in aids.iter().copied().enumerate() {
            let rows: Vec<(String, i64)> = self
                .raw_ref()
                .prepare("select crate, min(duration) from rustc_compilation where aid = ? group by crate")
                .unwrap()
                .query_map(params![&aid.0], |row| {
                    Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)?))
                })
                .unwrap()
                .map(|r| r.unwrap())
                .collect();
            for (krate, min_duration) in rows {
                let v = results
                    .entry(krate)
                    .or_insert_with(|| vec![None; aids.len()]);
                v[idx] = Some(Duration::from_nanos(min_duration as u64));
            }
        }

        results
    }

    async fn get_compile_pstats(
        &self,
        series: &[u32],
        artifact_row_ids: &[Option<ArtifactIdNumber>],
    ) -> Vec<Vec<Option<f64>>> {
        let mut conn = self.raw_ref();
        let tx = conn.transaction().unwrap();
        let mut query = tx
            .prepare_cached("select min(value) from pstat where series = ? and aid = ?;")
            .unwrap();
        series
            .iter()
            .map(|sid| {
                let elements = artifact_row_ids
                    .iter()
                    .map(|aid| {
                        aid.and_then(|aid| {
                            query
                                .query_row(params![&sid, &aid.0], |row| row.get(0))
                                .unwrap_or_else(|e| {
                                    panic!("{e:?}: series={sid:?}, aid={aid:?}");
                                })
                        })
                    })
                    .collect::<Vec<_>>();
                if elements.is_empty() {
                    vec![None; artifact_row_ids.len()]
                } else {
                    elements
                }
            })
            .collect()
    }

    async fn get_compile_pstats_all_samples(
        &self,
        series: &[u32],
        artifact_row_ids: &[Option<ArtifactIdNumber>],
    ) -> Vec<Vec<Vec<f64>>> {
        let conn = self.raw_ref();
        let mut query = conn
            .prepare_cached("select value from pstat where series = ? and aid = ?;")
            .unwrap();
        series
            .iter()
            .map(|sid| {
                artifact_row_ids
                    .iter()
                    .map(|aid| match aid {
                        Some(aid) => {
                            let values: Vec<f64> = query
                                .query_map(params![&sid, &aid.0], |row| row.get(0))
                                .unwrap_or_else(|e| {
                                    panic!("{e:?}: series={sid:?}, aid={aid:?}");
                                })
                                .filter_map(|r| r.ok())
                                .collect();
                            values
                        }
                        None => vec![],
                    })
                    .collect::<Vec<_>>()
            })
            .collect()
    }
    async fn get_runtime_pstats(
        &self,
        runtime_pstat_series_row_ids: &[u32],
        artifact_row_ids: &[Option<ArtifactIdNumber>],
    ) -> Vec<Vec<Option<f64>>> {
        let num_sids = runtime_pstat_series_row_ids.len();
        let num_aids = artifact_row_ids.len();
        let mut result = vec![vec![None; num_aids]; num_sids];

        let params = match BatchQueryParams::new(runtime_pstat_series_row_ids, artifact_row_ids) {
            Some(p) => p,
            None => return result,
        };

        let sql = format!(
            "SELECT series, aid, MIN(value) FROM runtime_pstat \
             WHERE series IN ({}) AND aid IN ({}) \
             GROUP BY series, aid",
            params.sid_placeholders, params.aid_placeholders
        );

        let conn = self.raw_ref();
        let mut stmt = conn.prepare_cached(&sql).unwrap();
        params.bind_params(&mut stmt, runtime_pstat_series_row_ids);

        let mut rows = stmt.raw_query();
        while let Some(row) = rows.next().unwrap() {
            let sid: u32 = row.get::<_, i64>(0).unwrap() as u32;
            let aid: u32 = row.get::<_, i64>(1).unwrap() as u32;
            let value: Option<f64> = row.get(2).unwrap();
            if let (Some(&si), Some(&ai)) = (params.sid_map.get(&sid), params.aid_map.get(&aid)) {
                result[si][ai] = value;
            }
        }

        result
    }

    async fn get_runtime_pstats_all_samples(
        &self,
        runtime_pstat_series_row_ids: &[u32],
        artifact_row_ids: &[Option<ArtifactIdNumber>],
    ) -> Vec<Vec<Vec<f64>>> {
        let num_sids = runtime_pstat_series_row_ids.len();
        let num_aids = artifact_row_ids.len();
        let mut result = vec![vec![vec![]; num_aids]; num_sids];

        let params = match BatchQueryParams::new(runtime_pstat_series_row_ids, artifact_row_ids) {
            Some(p) => p,
            None => return result,
        };

        let sql = format!(
            "SELECT series, aid, value FROM runtime_pstat \
             WHERE series IN ({}) AND aid IN ({}) AND value IS NOT NULL \
             ORDER BY series, aid",
            params.sid_placeholders, params.aid_placeholders
        );

        let conn = self.raw_ref();
        let mut stmt = conn.prepare_cached(&sql).unwrap();
        params.bind_params(&mut stmt, runtime_pstat_series_row_ids);

        let mut rows = stmt.raw_query();
        while let Some(row) = rows.next().unwrap() {
            let sid: u32 = row.get::<_, i64>(0).unwrap() as u32;
            let aid: u32 = row.get::<_, i64>(1).unwrap() as u32;
            let value: f64 = row.get(2).unwrap();
            if let (Some(&si), Some(&ai)) = (params.sid_map.get(&sid), params.aid_map.get(&aid)) {
                result[si][ai].push(value);
            }
        }

        result
    }

    async fn get_runtime_json_pstats(
        &self,
        runtime_pstat_series_row_ids: &[u32],
        artifact_row_ids: &[Option<ArtifactIdNumber>],
    ) -> Vec<Vec<Option<String>>> {
        let num_sids = runtime_pstat_series_row_ids.len();
        let num_aids = artifact_row_ids.len();
        let mut result: Vec<Vec<Option<String>>> = vec![vec![None; num_aids]; num_sids];

        let params = match BatchQueryParams::new(runtime_pstat_series_row_ids, artifact_row_ids) {
            Some(p) => p,
            None => return result,
        };

        let sql = format!(
            "SELECT series, aid, json_value FROM runtime_pstat \
             WHERE series IN ({}) AND aid IN ({}) AND json_value IS NOT NULL \
             ORDER BY series, aid, cid DESC",
            params.sid_placeholders, params.aid_placeholders
        );

        let conn = self.raw_ref();
        let mut stmt = conn.prepare_cached(&sql).unwrap();
        params.bind_params(&mut stmt, runtime_pstat_series_row_ids);

        let mut rows = stmt.raw_query();
        while let Some(row) = rows.next().unwrap() {
            let sid: u32 = row.get::<_, i64>(0).unwrap() as u32;
            let aid: u32 = row.get::<_, i64>(1).unwrap() as u32;
            if let (Some(&si), Some(&ai)) = (params.sid_map.get(&sid), params.aid_map.get(&aid)) {
                if result[si][ai].is_some() {
                    continue;
                }
                let json_val: String = row.get(2).unwrap();
                result[si][ai] = Some(json_val);
            }
        }

        result
    }

    async fn get_error(&self, aid: crate::ArtifactIdNumber) -> HashMap<String, String> {
        self.raw_ref()
            .prepare_cached("select context, message from error where aid = ?")
            .unwrap()
            .query_map(params![&aid.0], |row| Ok((row.get(0)?, row.get(1)?)))
            .unwrap()
            .collect::<Result<_, _>>()
            .unwrap()
    }

    async fn get_errors_for_job(&self, job_id: u32) -> Vec<(String, String)> {
        self.raw_ref()
            .prepare_cached("select context, message from error where job_id = ?")
            .unwrap()
            .query_map(params![&(job_id as i32)], |row| {
                Ok((row.get(0)?, row.get(1)?))
            })
            .unwrap()
            .collect::<Result<_, _>>()
            .unwrap()
    }

    async fn get_runtime_metrics_for_artifacts(
        &self,
        _aids: &[crate::ArtifactIdNumber],
    ) -> Vec<String> {
        // SQLite implementation not needed for production
        vec![]
    }

    async fn parent_of(&self, _sha: &str) -> Option<String> {
        None
    }

    async fn pr_of(&self, _sha: &str) -> Option<u32> {
        None
    }

    async fn list_master_commits_by_repo(&self) -> HashMap<String, Vec<crate::MasterCommitInfo>> {
        HashMap::new()
    }

    async fn list_self_profile(
        &self,
        aid: ArtifactId,
        crate_: &str,
        profile: &str,
        scenario: &str,
    ) -> Vec<(ArtifactIdNumber, CollectionId)> {
        self.raw_ref()
            .prepare(
                "select aid, cid from raw_self_profile where
        crate = ?1 and
        profile = ?2 and
        cache = ?3 and
        aid = (select id from artifact where name = ?4 and repo = ?5);",
            )
            .unwrap()
            .query_map(
                params![&crate_, profile, scenario, &aid.as_name(), &aid.repo()],
                |r| {
                    Ok((
                        ArtifactIdNumber(r.get::<_, i32>(0)? as u32),
                        CollectionId(r.get(1)?),
                    ))
                },
            )
            .unwrap()
            .collect::<Result<_, _>>()
            .unwrap()
    }

    async fn purge_artifact(&self, aid: &ArtifactId) {
        // Once we delete the artifact, all data associated with it should also be deleted
        // thanks to ON DELETE CASCADE.
        let info = aid.info();
        self.raw_ref()
            .execute("delete from artifact where name = ?1", [info.name])
            .unwrap();
    }

    async fn insert_benchmark_request(
        &self,
        _benchmark_request: &BenchmarkRequest,
    ) -> anyhow::Result<BenchmarkRequestInsertResult> {
        no_queue_implementation_abort!()
    }

    async fn load_benchmark_request_index(&self) -> anyhow::Result<BenchmarkRequestIndex> {
        Ok(BenchmarkRequestIndex {
            all: Default::default(),
        })
    }

    async fn load_pending_benchmark_requests(&self) -> anyhow::Result<PendingBenchmarkRequests> {
        no_queue_implementation_abort!()
    }

    async fn update_benchmark_request_status(
        &self,
        _tag: &str,
        _status: BenchmarkRequestStatus,
    ) -> anyhow::Result<()> {
        no_queue_implementation_abort!()
    }

    async fn reset_benchmark_request_for_rerun(&mut self, _tag: &str) -> anyhow::Result<bool> {
        no_queue_implementation_abort!()
    }

    async fn enqueue_benchmark_job(
        &self,
        _request_tag: &str,
        _benchmark_group: &str,
        _runtime_config: &QueueRuntimeConfig,
        _is_optional: bool,
        _tag: &str,
    ) -> JobEnqueueResult {
        no_queue_implementation_abort!()
    }

    async fn get_compile_test_cases_with_measurements(
        &self,
        artifact_row_id: &ArtifactIdNumber,
    ) -> anyhow::Result<HashSet<CompileTestCase>> {
        Ok(self
            .raw_ref()
            .prepare_cached(
                "SELECT DISTINCT crate, profile, scenario, backend, target
                FROM pstat_series
                WHERE id IN (
                    SELECT DISTINCT series
                    FROM pstat
                    WHERE aid = ?
                );",
            )?
            .query_map(params![artifact_row_id.0], |row| {
                Ok(CompileTestCase {
                    benchmark: Benchmark::from(row.get::<_, String>(0)?.as_str()),
                    profile: row.get::<_, String>(1)?.parse().unwrap(),
                    scenario: row.get::<_, String>(2)?.parse().unwrap(),
                    backend: row.get::<_, String>(3)?.parse().unwrap(),
                    target: row.get::<_, String>(4)?.parse().unwrap(),
                })
            })?
            .collect::<Result<_, _>>()?)
    }

    async fn get_runtime_benchmarks_with_measurements(
        &self,
        artifact_row_id: &ArtifactIdNumber,
    ) -> anyhow::Result<HashSet<RuntimeTestCase>> {
        Ok(self
            .raw_ref()
            .prepare_cached(
                "SELECT DISTINCT benchmark, target
                FROM runtime_pstat_series
                WHERE id IN (
                    SELECT DISTINCT series
                    FROM runtime_pstat
                    WHERE aid = ?
                );",
            )?
            .query_map(params![artifact_row_id.0], |row| {
                Ok(RuntimeTestCase {
                    benchmark: Benchmark::from(row.get::<_, String>(0)?.as_str()),
                    target: row.get::<_, String>(1)?.parse().unwrap(),
                })
            })?
            .collect::<Result<_, _>>()?)
    }

    async fn start_collector(
        &self,
        _collector_name: &str,
        _commit_sha: &str,
    ) -> anyhow::Result<Option<CollectorConfig>> {
        no_queue_implementation_abort!()
    }

    async fn dequeue_benchmark_job(
        &self,
        _collector_name: &str,
    ) -> anyhow::Result<Option<(BenchmarkJob, ArtifactId)>> {
        no_queue_implementation_abort!()
    }

    async fn add_collector_config(
        &self,
        _collector_name: &str,
        _is_active: bool,
    ) -> anyhow::Result<CollectorConfig> {
        no_queue_implementation_abort!()
    }

    async fn maybe_mark_benchmark_request_as_completed(&self, _tag: &str) -> anyhow::Result<bool> {
        no_queue_implementation_abort!()
    }

    async fn mark_benchmark_job_as_completed(
        &self,
        _id: u32,
        _conclusion: BenchmarkJobConclusion,
    ) -> anyhow::Result<()> {
        no_queue_implementation_abort!()
    }

    async fn get_jobs_of_in_progress_benchmark_requests(
        &self,
    ) -> anyhow::Result<HashMap<String, Vec<BenchmarkJob>>> {
        no_queue_implementation_abort!()
    }

    async fn get_jobs_of_benchmark_requests(
        &self,
        _tags: &[String],
    ) -> anyhow::Result<Vec<BenchmarkJob>> {
        no_queue_implementation_abort!()
    }

    async fn get_collector_configs(&self) -> anyhow::Result<Vec<CollectorConfig>> {
        no_queue_implementation_abort!()
    }

    async fn update_collector_heartbeat(&self, _collector_name: &str) -> anyhow::Result<()> {
        no_queue_implementation_abort!()
    }

    async fn get_last_n_completed_benchmark_requests(
        &self,
        _count: u64,
    ) -> anyhow::Result<Vec<BenchmarkRequestWithErrors>> {
        no_queue_implementation_abort!()
    }

    fn supports_job_queue(&self) -> bool {
        false
    }
}

fn parse_artifact_id(ty: &str, sha: &str, date: Option<i64>, repo: &str) -> ArtifactId {
    match ty {
        "master" => ArtifactId::Commit {
            repo: repo.to_owned(),
            commit: Commit {
                sha: sha.to_owned(),
                date: Date(
                    Utc.timestamp_opt(date.expect("master has date"), 0)
                        .unwrap(),
                ),
                r#type: CommitType::Master,
            },
        },
        "try" => ArtifactId::Commit {
            repo: repo.to_owned(),
            commit: Commit {
                sha: sha.to_owned(),
                date: date
                    .map(|d| Date(Utc.timestamp_opt(d, 0).unwrap()))
                    .unwrap_or_else(|| Date::ymd_hms(2000, 1, 1, 0, 0, 0)),
                r#type: CommitType::Try,
            },
        },
        "release" => ArtifactId::Tag {
            repo: repo.to_owned(),
            tag: sha.to_owned(),
        },
        _ => panic!("unknown artifact type: {ty:?}"),
    }
}