From 875c4be1158aaeb936af5f836dfacc7672d3a958 Mon Sep 17 00:00:00 2001 From: Bengineer Date: Thu, 19 Mar 2026 12:02:48 +0000 Subject: [PATCH 1/3] feat: Enhance Postgres sink with improved query handling and data structure updates - Introduced new queries for inserting tables, columns, and dead fields in the Postgres sink. - Refactored the `PgTable` struct to include methods for generating insert queries. - Updated the `TableUpgrade` struct to track upgraded columns and handle column renaming and type changes more effectively. - Implemented async functions for fetching tables, columns, and dead fields from the database. - Added a new `Executable` trait to streamline query execution across different database types. - Enhanced the `PgTypeDef` trait to support more flexible row fetching and mapping. - Updated the `FlexStr` and `QueryLike` structures to improve query handling. - Adjusted example configurations for testing with new schema and batch sizes. --- Cargo.lock | 4 + bins/torii-introspect-bin/src/main.rs | 1 - crates/dojo/migrations/001_dojo_store.sql | 22 +- crates/dojo/src/store/postgres.rs | 225 +++++++++--- crates/introspect-postgres-sink/Cargo.toml | 1 + .../migrations/003_store.sql | 65 ++++ crates/introspect-postgres-sink/src/create.rs | 9 +- .../introspect-postgres-sink/src/processor.rs | 167 ++++----- crates/introspect-postgres-sink/src/query.rs | 319 +++++++++++++++--- crates/introspect-postgres-sink/src/table.rs | 41 ++- .../introspect-postgres-sink/src/upgrade.rs | 8 +- crates/introspect/src/postgres/owned.rs | 76 ++--- crates/introspect/src/postgres/types.rs | 6 + crates/postgres/Cargo.toml | 3 + crates/postgres/src/db.rs | 18 +- crates/torii-common/Cargo.toml | 7 + crates/torii-common/src/lib.rs | 1 + crates/torii-common/src/sql.rs | 241 +++++++++++++ examples/introspect/restart.rs | 14 +- 19 files changed, 918 insertions(+), 310 deletions(-) create mode 100644 crates/introspect-postgres-sink/migrations/003_store.sql create mode 100644 crates/torii-common/src/sql.rs diff --git a/Cargo.lock b/Cargo.lock index 1de10c11..820f99fb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4640,9 +4640,11 @@ dependencies = [ "anyhow", "async-trait", "base64 0.22.1", + "itertools 0.14.0", "reqwest", "serde", "serde_json", + "sqlx", "starknet", "tokio", "tracing", @@ -4897,6 +4899,7 @@ dependencies = [ "thiserror 2.0.18", "tokio", "torii", + "torii-common", "torii-introspect", "torii-postgres", "tracing", @@ -4986,6 +4989,7 @@ dependencies = [ "thiserror 2.0.18", "tokio", "torii", + "torii-common", "tracing", "xxhash-rust", ] diff --git a/bins/torii-introspect-bin/src/main.rs b/bins/torii-introspect-bin/src/main.rs index d668b9b3..819143c5 100644 --- a/bins/torii-introspect-bin/src/main.rs +++ b/bins/torii-introspect-bin/src/main.rs @@ -429,7 +429,6 @@ async fn run_with_postgres( decoder.store.initialize().await?; sink.initialize_introspect_pg_sink().await?; decoder.load_tables(&[]).await?; - sink.load_tables_no_commit(decoder.get_tables()?)?; let decoder: Arc = Arc::new(decoder); diff --git a/crates/dojo/migrations/001_dojo_store.sql b/crates/dojo/migrations/001_dojo_store.sql index 3ccf6e3f..16cac3b4 100644 --- a/crates/dojo/migrations/001_dojo_store.sql +++ b/crates/dojo/migrations/001_dojo_store.sql @@ -11,18 +11,12 @@ BEGIN CREATE DOMAIN uint64 AS NUMERIC(20, 0) CHECK (VALUE >= 0 AND VALUE < power(2::numeric, 64)); END IF; - - IF to_regtype('introspect.attribute') IS NULL THEN - CREATE TYPE introspect.attribute AS ( - name TEXT, - data bytea - ); - END IF; END $$; CREATE TABLE IF NOT EXISTS dojo.tables ( owner felt252, id felt252 NOT NULL, + block_number uint64 NOT NULL, name TEXT NOT NULL, attributes TEXT[] NOT NULL DEFAULT '{}', keys felt252[] NOT NULL, @@ -30,21 +24,23 @@ CREATE TABLE IF NOT EXISTS dojo.tables ( legacy BOOLEAN NOT NULL, created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), - created_block uint64 NOT NULL, - updated_block uint64 NOT NULL, created_tx felt252 NOT NULL, updated_tx felt252 NOT NULL, - PRIMARY KEY (owner, id) + PRIMARY KEY (owner, id, block_number) ); CREATE TABLE IF NOT EXISTS dojo.columns( owner felt252, "table" felt252 NOT NULL , id felt252 NOT NULL, + block_number uint64 NOT NULL, name TEXT NOT NULL, - attributes introspect.attribute[] NOT NULL DEFAULT '{}', + attributes TEXT[] NOT NULL DEFAULT '{}', type_def jsonb NOT NULL, - PRIMARY KEY (owner, "table", id), - FOREIGN KEY (owner, "table") REFERENCES dojo.tables(owner, id) + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + created_tx felt252 NOT NULL, + updated_tx felt252 NOT NULL, + PRIMARY KEY (owner, "table", id, block_number) ); diff --git a/crates/dojo/src/store/postgres.rs b/crates/dojo/src/store/postgres.rs index ba368a26..2d22b00d 100644 --- a/crates/dojo/src/store/postgres.rs +++ b/crates/dojo/src/store/postgres.rs @@ -3,22 +3,23 @@ use crate::decoder::primary_field_def; use crate::table::DojoTableInfo; use crate::DojoTable; use async_trait::async_trait; -use introspect_types::{ColumnInfo, ResultInto}; +use introspect_types::{Attribute, ColumnInfo, ResultInto, TypeDef}; use itertools::Itertools; use sqlx::migrate::Migrator; use sqlx::postgres::PgArguments; use sqlx::query::Query; -use sqlx::{FromRow, PgPool, Postgres}; +use sqlx::types::Json; +use sqlx::{FromRow, Postgres}; use starknet_types_core::felt::Felt; use std::collections::HashMap; use std::ops::Deref; -use torii_introspect::postgres::owned::{column_info_insert_query, PgTypeDef}; +use torii_introspect::postgres::owned::PgTypeDef; use torii_introspect::postgres::{PgFelt, SqlxResult}; use torii_introspect::schema::ColumnKeyTrait; use torii_postgres::db::PostgresConnection; pub const FETCH_TABLES_QUERY: &str = r#" - SELECT + SELECT DISTINCT ON (owner, id) id, name, attributes, @@ -26,38 +27,64 @@ pub const FETCH_TABLES_QUERY: &str = r#" "values", legacy FROM dojo.tables - WHERE $1::felt252[] = '{}' OR owner = ANY($1)"#; + WHERE $1::felt252[] = '{}' OR owner = ANY($1) + ORDER BY owner, id, block_number DESC"#; pub const FETCH_COLUMNS_QUERY: &str = r#" - SELECT + SELECT DISTINCT ON (owner, "table", id) "table", id, name, attributes, type_def FROM dojo.columns - WHERE $1::felt252[] = '{}' OR owner = ANY($1)"#; + WHERE $1::felt252[] = '{}' OR owner = ANY($1) + ORDER BY owner, "table", id, block_number DESC"#; + +pub const FETCH_TABLES_BEFORE_QUERY: &str = r#" + SELECT DISTINCT ON (owner, id) + id, + name, + attributes, + keys, + "values", + legacy + FROM dojo.tables + WHERE ($1::felt252[] = '{}' OR owner = ANY($1)) AND block_number < $2::uint64 + ORDER BY owner, id, block_number DESC"#; + +pub const FETCH_COLUMNS_BEFORE_QUERY: &str = r#" + SELECT DISTINCT ON (owner, "table", id) + "table", + id, + name, + attributes, + type_def + FROM dojo.columns + WHERE ($1::felt252[] = '{}' OR owner = ANY($1)) AND block_number < $2::uint64 + ORDER BY owner, "table", id, block_number DESC"#; pub const INSERT_TABLE_QUERY: &str = r#" - INSERT INTO dojo.tables (owner, id, name, attributes, keys, "values", legacy, updated_at, created_block, updated_block, created_tx, updated_tx) - VALUES ($1, $2, $3, $4, $5, $6, $7, NOW(), $8::uint64, $8::uint64, $9, $9) - ON CONFLICT (owner, id) DO UPDATE SET + INSERT INTO dojo.tables (owner, id, block_number, name, attributes, keys, "values", legacy, updated_at, created_tx, updated_tx) + VALUES ($1, $2, $3::uint64, $4, $5, $6, $7, $8, NOW(), $9, $9) + ON CONFLICT (owner, id, block_number) DO UPDATE SET name = EXCLUDED.name, attributes = EXCLUDED.attributes, keys = EXCLUDED.keys, "values" = EXCLUDED."values", legacy = EXCLUDED.legacy, updated_at = NOW(), - updated_block = EXCLUDED.updated_block, updated_tx = EXCLUDED.updated_tx"#; pub const INSERT_COLUMN_QUERY: &str = r#" - INSERT INTO dojo.columns (owner, "table", id, name, attributes, type_def) - VALUES ($1, $2, $3, $4, $5, $6) - ON CONFLICT (owner, "table", id) DO UPDATE SET + INSERT INTO dojo.columns (owner, "table", id, block_number, name, attributes, type_def, updated_at, created_tx, updated_tx) + VALUES ($1, $2, $3, $4::uint64, $5, $6, $7, NOW(), $8, $8) + ON CONFLICT (owner, "table", id, block_number) DO UPDATE SET name = EXCLUDED.name, attributes = EXCLUDED.attributes, - type_def = EXCLUDED.type_def"#; + type_def = EXCLUDED.type_def, + updated_at = NOW(), + updated_tx = EXCLUDED.updated_tx"#; pub const DOJO_STORE_MIGRATIONS: Migrator = sqlx::migrate!(); @@ -89,7 +116,7 @@ impl DojoPgStoreError { } #[derive(FromRow)] -pub struct TableRow { +pub struct DojoTableRow { id: PgFelt, name: String, attributes: Vec, @@ -99,8 +126,17 @@ pub struct TableRow { legacy: bool, } -impl From for ((), DojoTable) { - fn from(value: TableRow) -> Self { +#[derive(FromRow)] +pub struct DojoColumnRow { + table: PgFelt, + id: PgFelt, + name: String, + attributes: Vec, + type_def: Json, +} + +impl From for ((), DojoTable) { + fn from(value: DojoTableRow) -> Self { ( (), DojoTable { @@ -117,8 +153,8 @@ impl From for ((), DojoTable) { } } -impl From for (Felt, DojoTableInfo) { - fn from(value: TableRow) -> Self { +impl From for (Felt, DojoTableInfo) { + fn from(value: DojoTableRow) -> Self { ( value.id.into(), DojoTableInfo { @@ -134,54 +170,121 @@ impl From for (Felt, DojoTableInfo) { } } -#[async_trait] -impl PgTypeDef for DojoTableInfo { - type Row = TableRow; - async fn get_rows( - pool: &PgPool, - query: &'static str, - owners: &[Felt], - ) -> SqlxResult> { - Self::get_pg_rows(pool, query, owners) - .await - .map(|rows| rows.into_iter().map_into().collect_vec()) +impl From for (K, ColumnInfo) +where + K: ColumnKeyTrait, +{ + fn from(value: DojoColumnRow) -> Self { + ( + K::from_parts(value.table.into(), value.id.into()), + ColumnInfo { + name: value.name, + attributes: value + .attributes + .into_iter() + .map(Attribute::new_empty) + .collect(), + type_def: value.type_def.0, + }, + ) } } -#[async_trait] -impl PgTypeDef<()> for DojoTable { - type Row = TableRow; - async fn get_rows( - pool: &PgPool, - query: &'static str, - owners: &[Felt], - ) -> SqlxResult> { - Self::get_pg_rows(pool, query, owners) - .await - .map(|rows| rows.into_iter().map_into().collect_vec()) - } -} +// #[async_trait] +// impl PgTypeDef for DojoTableInfo { +// type Row = DojoTableRow; +// async fn get_rows( +// pool: &PgPool, +// query: &'static str, +// owners: &[Felt], +// ) -> SqlxResult> { +// Self::get_pg_rows(pool, query, owners) +// .await +// .map(|rows| rows.into_iter().map_into().collect_vec()) +// } +// } + +// #[async_trait] +// impl PgTypeDef<()> for DojoTable { +// type Row = DojoTableRow; +// async fn get_rows( +// pool: &PgPool, +// query: &'static str, +// owners: &[Felt], +// ) -> SqlxResult> { +// Self::get_pg_rows(pool, query, owners) +// .await +// .map(|rows| rows.into_iter().map_into().collect_vec()) +// } +// } pub fn table_insert_query( owner: &Felt, id: &Felt, + block_number: u64, name: &str, attributes: &[String], keys: &[Felt], values: &[Felt], legacy: bool, - created_block: u64, created_tx: &Felt, ) -> Query<'static, Postgres, PgArguments> { sqlx::query::(INSERT_TABLE_QUERY) .bind(PgFelt::from(*owner)) .bind(PgFelt::from(*id)) + .bind(block_number.to_string()) .bind(name.to_owned()) .bind(attributes.to_owned()) .bind(keys.iter().copied().map(PgFelt::from).collect_vec()) .bind(values.iter().copied().map(PgFelt::from).collect_vec()) .bind(legacy) - .bind(created_block.to_string()) + .bind(PgFelt::from(*created_tx)) +} + +pub fn column_info_insert_query( + query: &'static str, + owner: &Felt, + table: &Felt, + id: &Felt, + block_number: u64, + info: &ColumnInfo, + created_tx: &Felt, +) -> Query<'static, Postgres, PgArguments> { + column_insert_query( + query, + owner, + table, + id, + block_number, + &info.name, + info.attributes + .iter() + .map(|s| s.name.to_string()) + .collect_vec(), + &info.type_def, + created_tx, + ) +} + +pub fn column_insert_query( + query: &'static str, + owner: &Felt, + table: &Felt, + id: &Felt, + block_number: u64, + name: &str, + attributes: Vec, + type_def: &TypeDef, + created_tx: &Felt, +) -> Query<'static, Postgres, PgArguments> { + sqlx::query::(query) + .bind(PgFelt::from(*owner)) + .bind(PgFelt::from(*table)) + .bind(PgFelt::from(*id)) + .bind(block_number.to_string()) + .bind(name.to_owned()) + .bind(attributes) + .bind(Json(type_def.clone())) .bind(PgFelt::from(*created_tx)) } @@ -195,12 +298,12 @@ impl DojoTable { table_insert_query( owner, &self.id, + block_number, &self.name, &self.attributes, &self.key_fields.iter().copied().collect_vec(), &self.value_fields.iter().copied().collect_vec(), self.legacy, - block_number, tx_hash, ) } @@ -246,22 +349,32 @@ impl DojoStoreTrait for PgStore Result, Self::Error> { - let mut tables = DojoTable::get_rows(self.pool(), FETCH_TABLES_QUERY, owners) - .await? - .into_iter() - .map(|row| row.1) - .collect_vec(); + let mut tables = + PgTypeDef::get_rows::(self.pool(), FETCH_TABLES_QUERY, owners) + .await? + .into_iter() + .map(|row: ((), DojoTable)| row.1) + .collect_vec(); let mut columns: HashMap<(Felt, Felt), _> = - ColumnInfo::get_hash_map(self.pool(), FETCH_COLUMNS_QUERY, owners).await?; + ColumnInfo::get_hash_map::(self.pool(), FETCH_COLUMNS_QUERY, owners) + .await?; for table in &mut tables { for key in table.key_fields.iter().chain(table.value_fields.iter()) { let column = columns.remove(&(table.id, *key)).ok_or_else(|| { diff --git a/crates/introspect-postgres-sink/Cargo.toml b/crates/introspect-postgres-sink/Cargo.toml index 1702540e..816f88dd 100644 --- a/crates/introspect-postgres-sink/Cargo.toml +++ b/crates/introspect-postgres-sink/Cargo.toml @@ -33,3 +33,4 @@ itertools.workspace = true # Local crates torii-introspect.workspace = true torii-postgres.workspace = true +torii-common.workspace = true diff --git a/crates/introspect-postgres-sink/migrations/003_store.sql b/crates/introspect-postgres-sink/migrations/003_store.sql new file mode 100644 index 00000000..d58e2145 --- /dev/null +++ b/crates/introspect-postgres-sink/migrations/003_store.sql @@ -0,0 +1,65 @@ +CREATE SCHEMA IF NOT EXISTS introspect; + +DO $$ +BEGIN + IF to_regtype('introspect.attribute') IS NULL THEN + CREATE TYPE introspect.attribute AS ( + name TEXT, + data bytea + ); + END IF; + + IF to_regtype('introspect.primary_def') IS NULL THEN + CREATE TYPE introspect.primary_def AS ( + name TEXT, + attributes introspect.attribute[], + type_def jsonb + ); + END IF; +END $$; + +CREATE TABLE IF NOT EXISTS introspect.db_tables ( + "schema" TEXT NOT NULL, + id felt252 NOT NULL, + name TEXT NOT NULL, + primary_def introspect.primary_def NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + created_block uint64 NOT NULL, + updated_block uint64 NOT NULL, + created_tx felt252 NOT NULL, + updated_tx felt252 NOT NULL, + PRIMARY KEY ("schema", id) +); + +CREATE TABLE IF NOT EXISTS introspect.db_columns( + "schema" TEXT NOT NULL, + "table" felt252 NOT NULL, + id felt252 NOT NULL, + name TEXT NOT NULL, + type_def JSONB NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + created_block uint64 NOT NULL, + updated_block uint64 NOT NULL, + created_tx felt252 NOT NULL, + updated_tx felt252 NOT NULL, + PRIMARY KEY ("schema", "table", id), + FOREIGN KEY ("schema", "table") REFERENCES introspect.db_tables("schema", id) +); + +CREATE TABLE IF NOT EXISTS introspect.db_dead_fields ( + "schema" TEXT NOT NULL, + "table" felt252, + id uint128, + name TEXT, + type_def JSONB, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + created_block uint64 NOT NULL, + updated_block uint64 NOT NULL, + created_tx felt252 NOT NULL, + updated_tx felt252 NOT NULL, + PRIMARY KEY ("schema", "table", id), + FOREIGN KEY ("schema", "table") REFERENCES introspect.db_tables("schema", id) +); \ No newline at end of file diff --git a/crates/introspect-postgres-sink/src/create.rs b/crates/introspect-postgres-sink/src/create.rs index 195f27de..41bf4d85 100644 --- a/crates/introspect-postgres-sink/src/create.rs +++ b/crates/introspect-postgres-sink/src/create.rs @@ -11,6 +11,7 @@ use introspect_types::{ use itertools::Itertools; use starknet_types_core::felt::Felt; use std::rc::Rc; +use torii_common::sql::{PgQuery, Queries}; use torii_introspect::schema::TableInfo; use xxhash_rust::xxh3::Xxh3; @@ -309,12 +310,12 @@ impl CreatePgTable { pg_types: creates, }) } - pub fn make_queries(&self, queries: &mut Vec) { + pub fn make_queries(&self, queries: &mut Vec) { for pg_type in &self.pg_types { - queries.push(pg_type.to_string()); + queries.add(pg_type.to_string()); } - queries.push(self.to_string()); - queries.push(format!( + queries.add(self.to_string()); + queries.add(format!( r#"CREATE TRIGGER set_timestamps BEFORE INSERT ON {} FOR EACH ROW EXECUTE FUNCTION introspect.set_default_timestamps();"#, self.name )); diff --git a/crates/introspect-postgres-sink/src/processor.rs b/crates/introspect-postgres-sink/src/processor.rs index b98cca3c..196a7ef1 100644 --- a/crates/introspect-postgres-sink/src/processor.rs +++ b/crates/introspect-postgres-sink/src/processor.rs @@ -1,11 +1,10 @@ use crate::json::PostgresJsonSerializer; -use crate::query::CreatePgTable; +use crate::query::{fetch_columns, fetch_dead_fields, fetch_tables, CreatePgTable}; use crate::table::{DeadField, PgTable}; use crate::{PgDbError, PgDbResult, PgSchema, INTROSPECT_PG_SINK_MIGRATIONS}; -use introspect_types::{ColumnInfo, ResultInto, TypeDef}; +use introspect_types::ColumnInfo; use serde_json::Serializer as JsonSerializer; -use sqlx::types::Json; -use sqlx::{FromRow, PgPool}; +use sqlx::PgPool; use starknet_types_core::felt::Felt; use std::collections::HashMap; use std::io::Write; @@ -13,15 +12,16 @@ use std::ops::Deref; use std::rc::Rc; use std::sync::RwLock; use torii::etl::envelope::MetaData; +use torii_common::sql::{PgQuery, Queries}; use torii_introspect::events::{IntrospectBody, IntrospectMsg}; -use torii_introspect::postgres::types::Uint128; -use torii_introspect::postgres::PgFelt; use torii_introspect::schema::TableSchema; use torii_introspect::InsertsFields; use torii_postgres::PostgresConnection; pub const COMMIT_CMD: &str = "--COMMIT"; -pub const DEAD_MEMBERS_TABLE: &str = "__torii_dead_fields"; +pub const DEAD_MEMBERS_TABLE: &str = "__introspect_dead_fields"; +pub const TABLES_TABLE: &str = "__introspect_tables"; +pub const COLUMNS_TABLE: &str = "__introspect_columns"; pub const METADATA_CONFLICTS: &str = "__updated_at = NOW(), __updated_block = EXCLUDED.__updated_block, __updated_tx = EXCLUDED.__updated_tx"; #[derive(Debug, Default)] @@ -87,29 +87,47 @@ impl PostgresTables { &self, schema: &Rc, to_table: impl Into, - queries: &mut Vec, + metadata: &MetaData, + queries: &mut Vec, ) -> PgDbResult<()> { let (id, table) = Into::::into(to_table).into(); self.assert_table_not_exists(&id, &table.name)?; CreatePgTable::new(schema, &id, &table)?.make_queries(queries); + let table = PgTable::new(schema, table, None); + table.insert_queries( + &id, + None, + metadata.block_number.unwrap_or_default(), + metadata.transaction_hash, + queries, + )?; let mut tables: std::sync::RwLockWriteGuard<'_, HashMap> = self.write()?; - tables.insert(id, PgTable::new(schema, table, None)); + tables.insert(id, table); Ok(()) } pub fn update_table( &self, to_table: impl Into, - queries: &mut Vec, + metadata: &MetaData, + queries: &mut Vec, ) -> PgDbResult<()> { + let tx_hash = &metadata.transaction_hash; + let block_number = metadata.block_number.unwrap_or_default(); let (id, table) = Into::::into(to_table).into(); let mut tables = self.write()?; let existing = tables .get_mut(&id) .ok_or_else(|| PgDbError::TableNotFound(id))?; let upgrades = existing.update_from_info(&id, &table)?; - upgrades.to_queries(&id, queries); - Ok(()) + upgrades.to_queries(&id, block_number, tx_hash, queries)?; + existing.insert_queries( + &id, + Some(&upgrades.columns_upgraded), + block_number, + metadata.transaction_hash, + queries, + ) } pub fn assert_table_not_exists(&self, id: &Felt, name: &str) -> PgDbResult<()> { @@ -138,7 +156,7 @@ impl PostgresTables { &self, event: &InsertsFields, context: &MetaData, - queries: &mut Vec, + queries: &mut Vec, ) -> PgDbResult<()> { let tables = self.read().unwrap(); let table = match tables.get(&event.table) { @@ -178,7 +196,7 @@ impl PostgresTables { .unwrap(); } let string = unsafe { String::from_utf8_unchecked(writer) }; - queries.push(string); + queries.add(string); Ok(()) } @@ -187,11 +205,15 @@ impl PostgresTables { schema: &Rc, msg: &IntrospectMsg, metadata: &MetaData, - queries: &mut Vec, + queries: &mut Vec, ) -> PgDbResult<()> { match msg { - IntrospectMsg::CreateTable(event) => self.create_table(schema, event.clone(), queries), - IntrospectMsg::UpdateTable(event) => self.update_table(event.clone(), queries), + IntrospectMsg::CreateTable(event) => { + self.create_table(schema, event.clone(), metadata, queries) + } + IntrospectMsg::UpdateTable(event) => { + self.update_table(event.clone(), metadata, queries) + } IntrospectMsg::AddColumns(event) => self.set_table_dead(&event.table), IntrospectMsg::DropColumns(event) => self.set_table_dead(&event.table), IntrospectMsg::RetypeColumns(event) => self.set_table_dead(&event.table), @@ -206,48 +228,14 @@ impl PostgresTables { } } -fn make_dead_members_table_query(schema: &PgSchema) -> String { - format!( - r#"CREATE TABLE IF NOT EXISTS "{schema}"."{DEAD_MEMBERS_TABLE}" ( - "table" public.felt252, - id public.uint128, - name TEXT, - type_def JSONB, - PRIMARY KEY ("table", id) - )"#, - ) -} - fn make_schema_query(schema: &PgSchema) -> String { format!(r#"CREATE SCHEMA IF NOT EXISTS "{schema}""#) } -#[derive(Debug, FromRow)] -pub struct DeadFieldRow { - pub table: PgFelt, - pub id: Uint128, - pub name: String, - pub type_def: Json, -} - -impl From for (Felt, u128, DeadField) { - fn from(value: DeadFieldRow) -> Self { - ( - value.table.into(), - value.id.into(), - DeadField { - name: value.name, - type_def: value.type_def.0, - }, - ) - } -} - pub struct IntrospectPgDb { tables: PostgresTables, schema: PgSchema, pool: T, - dead_fields: DeadFields, } impl PostgresConnection for IntrospectPgDb { @@ -262,60 +250,36 @@ impl IntrospectPgDb { tables: PostgresTables::default(), schema: schema.into(), pool, - dead_fields: DeadFields::default(), } } - pub async fn read_dead_field_rows(&self) -> PgDbResult> { - let schema = &self.schema; - let query = - format!(r#"SELECT "table", id, name, type_def FROM "{schema}"."{DEAD_MEMBERS_TABLE}""#); - sqlx::query_as::<_, DeadFieldRow>(&query) - .fetch_all(self.pool()) - .await - .err_into() - } - - pub async fn load_dead_fields(&self) -> PgDbResult<()> { - let mut dead_fields_map: HashMap> = HashMap::new(); - for row in self.read_dead_field_rows().await? { - let (table, id, dead_field) = row.into(); - dead_fields_map - .entry(table) - .or_default() - .push((id, dead_field)); + pub async fn load_store_data(&self) -> PgDbResult<()> { + let mut tables = fetch_tables(self.pool(), &self.schema) + .await? + .into_iter() + .map(|t| t.to_table(&self.schema)) + .collect::>(); + for (table_id, id, column_info) in fetch_columns(self.pool(), &self.schema).await? { + if let Some(table) = tables.get_mut(&table_id) { + table.columns.insert(id, column_info); + } } - let mut tables = self.tables.write()?; - let mut dead_fields = self.dead_fields.write()?; - for (table, fields) in dead_fields_map { - match tables.get_mut(&table) { - Some(table) => table.dead.extend(fields), - None => dead_fields.entry(table).or_default().extend(fields), + for (table_id, id, field) in fetch_dead_fields(self.pool(), &self.schema).await? { + if let Some(table) = tables.get_mut(&table_id) { + table.dead.insert(id, field); } } + let mut tables_map = self.tables.write()?; + tables_map.extend(tables); Ok(()) } pub async fn initialize_introspect_pg_sink(&self) -> PgDbResult<()> { self.migrate(Some("introspect"), INTROSPECT_PG_SINK_MIGRATIONS) .await?; - let queries = [ - make_schema_query(&self.schema), - make_dead_members_table_query(&self.schema), - ]; - self.execute_queries(&queries).await?; - self.load_dead_fields().await - } - - pub fn load_tables_no_commit(&self, table_schemas: Vec) -> PgDbResult<()> { - let mut tables: std::sync::RwLockWriteGuard<'_, HashMap> = - self.tables.write()?; - let mut deads = self.dead_fields.write()?; - for table in table_schemas { - let (id, table) = table.into(); - tables.insert(id, PgTable::new(&self.schema, table, deads.remove(&id))); - } - Ok(()) + self.execute_queries(make_schema_query(&self.schema)) + .await?; + self.load_store_data().await } pub async fn process_message( @@ -329,7 +293,7 @@ impl IntrospectPgDb { self.tables .handle_message(&schema, msg, metadata, &mut queries)?; } - self.execute_queries(&queries).await?; + self.execute_queries(queries).await?; Ok(()) } @@ -349,14 +313,17 @@ impl IntrospectPgDb { ); } } - let mut start = 0; - for (n, query) in queries.iter().enumerate() { - if query == COMMIT_CMD { - self.execute_queries(&queries[start..n]).await?; - start = n + 1; + let mut batch = Vec::new(); + for query in queries { + if query == *COMMIT_CMD { + self.execute_queries(std::mem::take(&mut batch)).await?; + } else { + batch.push(query); } } - self.execute_queries(&queries[start..]).await?; + if !batch.is_empty() { + self.execute_queries(batch).await?; + } Ok(results) } } diff --git a/crates/introspect-postgres-sink/src/query.rs b/crates/introspect-postgres-sink/src/query.rs index 0f6cb98e..66e67e87 100644 --- a/crates/introspect-postgres-sink/src/query.rs +++ b/crates/introspect-postgres-sink/src/query.rs @@ -1,18 +1,79 @@ -use introspect_types::{MemberDef, TypeDef}; +use introspect_types::{ColumnDef, ColumnInfo, MemberDef, PrimaryDef, TypeDef}; +use itertools::Itertools; +use sqlx::error::BoxDynError; +use sqlx::prelude::FromRow; +use sqlx::Error::Encode as EncodeError; +use sqlx::{postgres::PgArguments, types::Json}; +use sqlx::{Arguments, Executor, Postgres}; use starknet_types_core::felt::Felt; +use torii_common::sql::{PgQuery, Queries}; +use torii_introspect::postgres::types::{PgPrimary, Uint128}; use torii_introspect::postgres::PgFelt; +use torii_postgres::SqlxResult; +use crate::table::PgTable; use crate::{ - processor::{COMMIT_CMD, DEAD_MEMBERS_TABLE}, - table::DeadField, - PgSchema, PostgresField, PostgresType, PrimaryKey, SchemaName, + processor::COMMIT_CMD, table::DeadField, PgSchema, PostgresField, PostgresType, PrimaryKey, + SchemaName, }; +use std::collections::HashMap; use std::{ fmt::{Display, Formatter, Result as FmtResult, Write}, rc::Rc, }; const CREATE_METADATA_COLUMNS: &str = "__created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), __updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), __created_block public.uint64 NOT NULL, __updated_block public.uint64 NOT NULL, __created_tx public.felt252 NOT NULL, __updated_tx public.felt252 NOT NULL);"; +const INSERT_DEAD_MEMBER_QUERY: &str = r#"INSERT INTO introspect.db_dead_fields + ("schema", "table", id, name, type_def, updated_at, created_block, updated_block, created_tx, updated_tx) + SELECT $1, $2, unnest($3::bigint[]), unnest($4::text[]), unnest($5::jsonb[]), NOW(), $6::uint64, $6::uint64, $7, $7 + ON CONFLICT ("schema", "table", id) DO UPDATE SET + name = EXCLUDED.name, type_def = EXCLUDED.type_def, updated_at = NOW(), updated_block = EXCLUDED.updated_block, updated_tx = EXCLUDED.updated_tx"#; +const INSERT_TABLE_QUERY: &str = r#"INSERT INTO introspect.db_tables + ("schema", id, name, primary_def, updated_at, created_block, updated_block, created_tx, updated_tx) + VALUES ($1, $2, $3, $4, NOW(), $5::uint64, $5::uint64, $6, $6) + ON CONFLICT ("schema", id) DO UPDATE SET + name = EXCLUDED.name, primary_def = EXCLUDED.primary_def, updated_at = NOW(), updated_block = EXCLUDED.updated_block, updated_tx = EXCLUDED.updated_tx"#; +const INSERT_COLUMN_QUERY: &str = r#"INSERT INTO introspect.db_columns + ("schema", "table", id, name, type_def, updated_at, created_block, updated_block, created_tx, updated_tx) + SELECT $1, $2, unnest($3::felt252[]), unnest($4::text[]), unnest($5::jsonb[]), NOW(), $6::uint64, $6::uint64, $7, $7 + ON CONFLICT ("schema", "table", id) DO UPDATE SET + name = EXCLUDED.name, type_def = EXCLUDED.type_def, updated_at = NOW(), updated_block = EXCLUDED.updated_block, updated_tx = EXCLUDED.updated_tx"#; + +const FETCH_TABLES_QUERY: &str = + r#"SELECT id, name, primary_def FROM introspect.db_tables WHERE "schema" = $1"#; +const FETCH_COLUMNS_QUERY: &str = + r#"SELECT "table", id, name, type_def FROM introspect.db_columns WHERE "schema" = $1"#; +const FETCH_DEAD_FIELDS_QUERY: &str = + r#"SELECT "table", id, name, type_def FROM introspect.db_dead_fields WHERE "schema" = $1"#; + +#[derive(FromRow)] +pub struct TableRow { + pub id: PgFelt, + pub name: String, + pub primary_def: PgPrimary, +} + +#[derive(FromRow)] +pub struct ColumnRow { + pub table: PgFelt, + pub id: PgFelt, + pub name: String, + pub type_def: Json, +} + +#[derive(FromRow)] +pub struct DeadFieldRow { + pub table: PgFelt, + pub id: Uint128, + pub name: String, + pub type_def: Json, +} + +pub struct PgTableHead { + id: Felt, + name: String, + primary: PrimaryDef, +} #[derive(Debug)] pub struct CreatePgTable { @@ -30,6 +91,7 @@ pub struct TableUpgrade { pub atomic: Vec, pub alters: Vec, pub columns: Vec, + pub columns_upgraded: Vec, pub dead: Vec, pub col_alters: Vec, } @@ -40,6 +102,7 @@ pub struct ColumnUpgrade { pub alters: Vec, pub dead: Vec, pub altered: bool, + pub upgraded: bool, } #[derive(Debug)] @@ -209,17 +272,20 @@ impl TableUpgrade { name: name.into(), old_name: None, columns: Vec::new(), + columns_upgraded: Vec::new(), alters: Vec::new(), atomic: Vec::new(), dead: Vec::new(), col_alters: Vec::new(), } } - pub fn rename_column(&mut self, old: &mut String, new: &str) { - if old != new { + pub fn rename_column(&mut self, old: &mut String, new: &str) -> bool { + let renamed = old != new; + if renamed { let old = std::mem::replace(old, new.to_string()); self.columns.push(ColumnMod::Rename(old, new.to_string())); } + renamed } pub fn rename_table(&mut self, new: &str) { if self.name != new { @@ -233,46 +299,65 @@ impl TableUpgrade { } pub fn retype_column( &mut self, - name: &str, + column: &ColumnDef, pg_type: Option, upgrade: ColumnUpgrade, field: PostgresType, ) { if let Some(pg_type) = pg_type { - self.columns.push(ColumnMod::Alter(pg_type.to_field(name))); + self.columns + .push(ColumnMod::Alter(pg_type.to_field(&column.name))); } if upgrade.altered { - self.col_alters.push(field.to_field(name)); + self.col_alters.push(field.to_field(&column.name)); + } + if upgrade.upgraded { + self.columns_upgraded.push(column.id); } self.atomic = upgrade.atomic; self.alters = upgrade.alters; self.dead = upgrade.dead; } - pub fn add_column(&mut self, name: &str, pg_type: PostgresType) { + pub fn add_column(&mut self, id: Felt, name: &str, pg_type: PostgresType) { self.columns.push(ColumnMod::Add(PostgresField::new( name.to_string(), pg_type, ))); + self.columns_upgraded.push(id); } - pub fn column_upgrade(&mut self) -> ColumnUpgrade { + pub fn column_upgrade(&mut self, upgraded: bool) -> ColumnUpgrade { ColumnUpgrade { atomic: std::mem::take(&mut self.atomic), alters: std::mem::take(&mut self.alters), dead: std::mem::take(&mut self.dead), altered: false, + upgraded, } } - pub fn to_queries(&self, table_id: &Felt, queries: &mut Vec) { + pub fn to_queries( + &self, + table_id: &Felt, + block_number: u64, + transaction_hash: &Felt, + queries: &mut Vec, + ) -> SqlxResult<()> { let schema = &self.schema; let name = &self.name; - if let Some(query) = insert_dead_member_query(&self.schema, table_id, &self.dead) { - queries.push(query); - } + queries.add( + insert_dead_member_query( + &self.schema, + table_id, + &self.dead, + block_number, + transaction_hash, + ) + .map_err(EncodeError)?, + ); if let Some(old_name) = &self.old_name { - queries.push(format!( + queries.add(format!( r#"ALTER TABLE "{schema}"."{old_name}" RENAME TO "{name}";"# )); } @@ -283,12 +368,13 @@ impl TableUpgrade { .iter() .for_each(|m| write!(alterations, "{m}, ").unwrap()); write!(alterations, "{last};").unwrap(); - queries.push(alterations); + queries.add(alterations); } self.alter_queries(queries); + Ok(()) } - fn alter_queries(&self, queries: &mut Vec) { + fn alter_queries(&self, queries: &mut Vec) { if let Some((last, others)) = self.col_alters.split_last() { let (schema, name) = (&self.schema, &self.name); let mut forward = format!(r#"ALTER TABLE "{schema}"."{name}" "#); @@ -316,9 +402,9 @@ impl TableUpgrade { r#"ALTER COLUMN "{col}" TYPE {pg_type} USING jsonb_populate_record(null::{pg_type}, "{col}");"# ) .unwrap(); - queries.push(forward); - self.alters.iter().for_each(|a| queries.push(a.to_string())); - queries.push(reverse); + queries.add(forward); + self.alters.iter().for_each(|a| queries.add(a.to_string())); + queries.add(reverse); } } } @@ -336,11 +422,11 @@ impl Display for ColumnMod { } impl StructUpgrade { - fn to_queries(&self, queries: &mut Vec) { + fn to_queries(&self, queries: &mut Vec) { let name = &self.name; self.mods .iter() - .for_each(|m| queries.push(format!("ALTER TYPE {name} {m};"))); + .for_each(|m| queries.add(format!("ALTER TYPE {name} {m};"))); } } @@ -364,26 +450,26 @@ impl Display for StructAlter { } impl EnumUpgrade { - fn to_queries(&self, queries: &mut Vec) { + fn to_queries(&self, queries: &mut Vec) { let name = &self.name; for (old, new) in &self.rename { - queries.push(format!( + queries.add(format!( r#"ALTER TYPE {name} RENAME VALUE '{old}' TO '{new}';"# )); } for variant in &self.add { - queries.push(format!(r#"ALTER TYPE {name} ADD VALUE '{variant}';"#)); + queries.add(format!(r#"ALTER TYPE {name} ADD VALUE '{variant}';"#)); } - queries.push(COMMIT_CMD.to_string()); + queries.add(COMMIT_CMD); } } impl TypeMod { - fn to_queries(&self, queries: &mut Vec) { + fn to_queries(&self, queries: &mut Vec) { match self { TypeMod::Struct(upgrade) => upgrade.to_queries(queries), TypeMod::Enum(upgrade) => upgrade.to_queries(queries), - TypeMod::Create(create) => queries.push(create.to_string()), + TypeMod::Create(create) => queries.add(create.to_string()), } } } @@ -422,6 +508,7 @@ impl ColumnUpgrade { field: field.to_string(), pg_type, }); + self.upgraded = true; } } @@ -432,6 +519,7 @@ impl ColumnUpgrade { mods: Vec, ) { if !mods.is_empty() { + self.upgraded = true; self.atomic.push(TypeMod::Struct(StructUpgrade { name: SchemaName::new(schema, name), mods, @@ -446,6 +534,8 @@ impl ColumnUpgrade { add: Vec, ) { if !rename.is_empty() || !add.is_empty() { + self.upgraded = true; + self.atomic.push(TypeMod::Enum(EnumUpgrade { name: SchemaName::new(schema, name), rename, @@ -454,6 +544,7 @@ impl ColumnUpgrade { } } pub fn add_dead_member(&mut self, id: u128, member: &MemberDef) { + self.upgraded = true; self.dead.push(DeadFieldWithId { id, name: member.name.clone(), @@ -487,35 +578,151 @@ impl From for TypeMod { } } +impl From for (Felt, Felt, ColumnInfo) { + fn from(value: ColumnRow) -> Self { + ( + value.table.into(), + value.id.into(), + ColumnInfo { + name: value.name, + attributes: Vec::new(), + type_def: value.type_def.0, + }, + ) + } +} + +impl From for (Felt, u128, DeadField) { + fn from(value: DeadFieldRow) -> Self { + ( + value.table.into(), + value.id.into(), + DeadField { + name: value.name, + type_def: value.type_def.0, + }, + ) + } +} + +impl From for PgTableHead { + fn from(value: TableRow) -> Self { + let row = value; + PgTableHead { + id: row.id.into(), + name: row.name, + primary: row.primary_def.into(), + } + } +} + +impl PgTableHead { + pub fn to_table(self, schema: &PgSchema) -> (Felt, PgTable) { + ( + self.id, + PgTable { + schema: schema.clone(), + name: self.name, + primary: self.primary, + columns: HashMap::new(), + alive: true, + dead: HashMap::new(), + }, + ) + } +} + fn insert_dead_member_query( schema: &PgSchema, table: &Felt, fields: &[DeadFieldWithId], -) -> Option { - fn write(w: &mut String, table: &PgFelt, field: &DeadFieldWithId) { - let type_def = serde_json::to_string(&field.type_def).unwrap(); - let name = field.name.replace('\'', "''"); - write!( - w, - "({table}, {id}, '{name}', '{type_def}'::jsonb)", - id = field.id, - ) - .unwrap(); - } - let table: PgFelt = (*table).into(); - let (last, rest) = fields.split_last()?; - let mut query = format!( - r#"INSERT INTO "{schema}"."{DEAD_MEMBERS_TABLE}" ("table", id, name, type_def) VALUES "# - ); - - for field in rest { - write(&mut query, &table, field); - query.push_str(", "); - } - write(&mut query, &table, last); - query - .push_str( - r#" ON CONFLICT ("table", id) DO UPDATE SET name = EXCLUDED.name, type_def = EXCLUDED.type_def"#, - ); - Some(query) + block_number: u64, + transaction_hash: &Felt, +) -> Result { + let mut args = PgArguments::default(); + args.add(schema.to_string())?; + args.add(PgFelt::from(*table))?; + args.add(fields.iter().map(|f| f.id.to_string()).collect::>())?; + args.add(fields.iter().map(|f| f.name.clone()).collect::>())?; + args.add(fields.iter().map(|f| Json(&f.type_def)).collect::>())?; + args.add(block_number.to_string())?; + args.add(PgFelt::from(*transaction_hash))?; + + Ok(PgQuery::new(INSERT_DEAD_MEMBER_QUERY, args)) +} + +pub fn insert_columns_query( + schema: &PgSchema, + table: &Felt, + columns: Vec<(&Felt, &ColumnInfo)>, + block_number: u64, + transaction_hash: &Felt, +) -> Result { + let mut args = PgArguments::default(); + args.add(schema.to_string())?; + args.add(PgFelt::from(*table))?; + args.add( + columns + .iter() + .map(|(id, _)| PgFelt::from(*id)) + .collect_vec(), + )?; + args.add(columns.iter().map(|(_, c)| c.name.clone()).collect_vec())?; + args.add(columns.iter().map(|(_, c)| Json(&c.type_def)).collect_vec())?; + args.add(block_number.to_string())?; + args.add(PgFelt::from(*transaction_hash))?; + + Ok(PgQuery::new(INSERT_COLUMN_QUERY, args)) +} + +pub fn insert_table_query( + schema: &PgSchema, + id: &Felt, + name: &str, + primary_def: &PrimaryDef, + block_number: u64, + transaction_hash: &Felt, +) -> Result { + let mut args = PgArguments::default(); + args.add(schema.to_string())?; + args.add(PgFelt::from(*id))?; + args.add(name.to_owned())?; + args.add(PgPrimary::from(primary_def))?; + args.add(block_number.to_string())?; + args.add(PgFelt::from(*transaction_hash))?; + + Ok(PgQuery::new(INSERT_TABLE_QUERY, args)) +} + +pub async fn fetch_tables<'e, 'c, E: 'e + Executor<'c, Database = Postgres>>( + conn: E, + schema: &PgSchema, +) -> SqlxResult> { + sqlx::query_as::<_, TableRow>(FETCH_TABLES_QUERY) + .bind(schema.to_string()) + .fetch_all(conn) + .await + .map(|rows| rows.into_iter().map_into().collect()) +} + +pub async fn fetch_columns<'e, 'c, E: 'e + Executor<'c, Database = Postgres>>( + conn: E, + schema: &PgSchema, +) -> SqlxResult> { + sqlx::query_as::<_, ColumnRow>(FETCH_COLUMNS_QUERY) + .bind(schema.to_string()) + .fetch_all(conn) + .await + .map(|rows| rows.into_iter().map_into().collect()) +} + +pub async fn fetch_dead_fields<'e, 'c, E: 'e + Executor<'c, Database = Postgres>>( + conn: E, + schema: &PgSchema, +) -> SqlxResult> { + sqlx::query_as::<_, DeadFieldRow>(FETCH_DEAD_FIELDS_QUERY) + .bind(schema.to_string()) + .fetch_all(conn) + .await + .map(|rows| rows.into_iter().map_into().collect()) } diff --git a/crates/introspect-postgres-sink/src/table.rs b/crates/introspect-postgres-sink/src/table.rs index 82683b08..eeef1a9c 100644 --- a/crates/introspect-postgres-sink/src/table.rs +++ b/crates/introspect-postgres-sink/src/table.rs @@ -1,10 +1,15 @@ -use introspect_types::{ColumnInfo, FeltIds, MemberDef, PrimaryDef, TypeDef}; +use introspect_types::{ColumnInfo, MemberDef, PrimaryDef, TypeDef}; use itertools::Itertools; +use sqlx::Error::Encode as EncodeError; use starknet_types_core::felt::Felt; use std::collections::HashMap; +use torii_common::sql::PgQuery; use torii_introspect::{schema::TableInfo, tables::RecordSchema}; -use crate::{PgSchema, PgTableError, TableResult}; +use crate::{ + query::{insert_columns_query, insert_table_query}, + PgDbResult, PgSchema, PgTableError, TableResult, +}; #[derive(Debug)] pub struct PgTable { @@ -12,7 +17,6 @@ pub struct PgTable { pub name: String, pub primary: PrimaryDef, pub columns: HashMap, - pub order: Vec, pub alive: bool, pub dead: HashMap, } @@ -56,13 +60,11 @@ impl PgTable { } pub fn new(schema: &PgSchema, info: TableInfo, dead: Option>) -> Self { - let order = info.columns.ids(); PgTable { schema: schema.clone(), name: info.name, primary: info.primary, columns: info.columns.into_iter().map_into().collect(), - order, alive: true, dead: dead.unwrap_or_default().into_iter().collect(), } @@ -70,4 +72,33 @@ impl PgTable { pub fn get_record_schema(&self, columns: &[Felt]) -> TableResult> { Ok(RecordSchema::new(&self.primary, self.columns(columns)?)) } + pub fn insert_queries( + &self, + id: &Felt, + column_ids: Option<&[Felt]>, + block_number: u64, + transaction_hash: Felt, + queries: &mut Vec, + ) -> PgDbResult<()> { + queries.push( + insert_table_query( + &self.schema, + id, + &self.name, + &self.primary, + block_number, + &transaction_hash, + ) + .map_err(EncodeError)?, + ); + let columns = match column_ids { + Some(ids) => ids.iter().zip(self.columns(ids)?).collect_vec(), + None => self.columns.iter().collect_vec(), + }; + queries.push( + insert_columns_query(&self.schema, id, columns, block_number, &transaction_hash) + .map_err(EncodeError)?, + ); + Ok(()) + } } diff --git a/crates/introspect-postgres-sink/src/upgrade.rs b/crates/introspect-postgres-sink/src/upgrade.rs index 50985905..f565d01d 100644 --- a/crates/introspect-postgres-sink/src/upgrade.rs +++ b/crates/introspect-postgres-sink/src/upgrade.rs @@ -37,8 +37,8 @@ impl PgTable { for column in columns { let branch = branch.branch(&column.id); if let Some(current) = self.columns.get_mut(&column.id) { - table_mod.rename_column(&mut current.name, &column.name); - let mut column_upgrade = table_mod.column_upgrade(); + let upgraded = table_mod.rename_column(&mut current.name, &column.name); + let mut column_upgrade = table_mod.column_upgrade(upgraded); let pg_type = current .type_def .compare_type( @@ -50,7 +50,7 @@ impl PgTable { ) .to_table_result(&self.name, ¤t.name)?; let type_name = current.type_def.get_pg_type(&schema, &branch)?; - table_mod.retype_column(¤t.name, pg_type, column_upgrade, type_name); + table_mod.retype_column(column, pg_type, column_upgrade, type_name); } else { let (column_id, info) = column.clone().into(); self.columns.insert(column_id, info); @@ -58,7 +58,7 @@ impl PgTable { column .type_def .extract_type(&schema, &branch, &mut table_mod.atomic)?; - table_mod.add_column(&column.name, pg_type); + table_mod.add_column(column_id, &column.name, pg_type); } } Ok(table_mod) diff --git a/crates/introspect/src/postgres/owned.rs b/crates/introspect/src/postgres/owned.rs index acf2c796..d88bd569 100644 --- a/crates/introspect/src/postgres/owned.rs +++ b/crates/introspect/src/postgres/owned.rs @@ -122,95 +122,67 @@ impl From for (Felt, Table) { } #[async_trait] -pub trait PgTypeDef +pub trait PgTypeRow where Self: Sized, - Self::Row: for<'r> FromRow<'r, PgRow> + Send + Unpin, { - type Row; - async fn get_rows( - pool: &PgPool, - query: &'static str, - owners: &[Felt], - ) -> SqlxResult>; async fn get_pg_rows( pool: &PgPool, query: &'static str, owners: &[Felt], - ) -> SqlxResult> { - sqlx::query_as(query) - .bind(owners.iter().copied().map(PgFelt::from).collect_vec()) - .fetch_all(pool) - .await - } - async fn get_hash_map( - pool: &PgPool, - query: &'static str, - owners: &[Felt], - ) -> SqlxResult> - where - Key: std::hash::Hash + Eq, - { - Self::get_rows(pool, query, owners) - .await - .map(|rows| rows.into_iter().collect()) - } + ) -> SqlxResult>; } #[async_trait] -impl PgTypeDef for ColumnInfo +impl PgTypeRow for T where - K: ColumnKeyTrait + Send + Sync, + Self: for<'r> FromRow<'r, PgRow> + Send + Unpin + Sized, { - type Row = ColumnRow; - async fn get_rows( + async fn get_pg_rows( pool: &PgPool, query: &'static str, owners: &[Felt], - ) -> SqlxResult> - where - Self: Sized, - { - ColumnDef::get_pg_rows(pool, query, owners) + ) -> SqlxResult> { + sqlx::query_as(query) + .bind(owners.iter().copied().map(PgFelt::from).collect_vec()) + .fetch_all(pool) .await - .map(|rows| rows.into_iter().map_into().collect_vec()) } } #[async_trait] -impl PgTypeDef for ColumnDef { - type Row = ColumnRow; - async fn get_rows( +pub trait PgTypeDef { + async fn get_rows( pool: &PgPool, query: &'static str, owners: &[Felt], - ) -> SqlxResult> + ) -> SqlxResult> where Self: Sized, + Row: PgTypeRow + Into<(Key, Self)>, { - Self::get_pg_rows(pool, query, owners) - .await - .map(|rows| rows.into_iter().map_into().collect_vec()) + let pg_rows: Vec = ::get_pg_rows(pool, query, owners).await?; + Ok(pg_rows.into_iter().map_into().collect_vec()) } -} - -#[async_trait] -impl PgTypeDef for Table { - type Row = TableRow; - async fn get_rows( + async fn get_hash_map( pool: &PgPool, query: &'static str, owners: &[Felt], - ) -> SqlxResult> + ) -> SqlxResult> where Self: Sized, + Key: std::hash::Hash + Eq, + Row: PgTypeRow + Into<(Key, Self)>, { - Self::get_pg_rows(pool, query, owners) + Self::get_rows::(pool, query, owners) .await - .map(|rows| rows.into_iter().map_into().collect_vec()) + .map(|rows| rows.into_iter().collect()) } } +#[async_trait] +impl PgTypeDef for T {} + pub trait PgColumnTrait { fn insert_query( &self, diff --git a/crates/introspect/src/postgres/types.rs b/crates/introspect/src/postgres/types.rs index 0c6cd869..b06dced8 100644 --- a/crates/introspect/src/postgres/types.rs +++ b/crates/introspect/src/postgres/types.rs @@ -149,6 +149,12 @@ impl From for PgFelt { } } +impl From<&Felt> for PgFelt { + fn from(value: &Felt) -> Self { + PgFelt(value.to_bytes_be()) + } +} + #[derive(sqlx::Type, Debug)] #[sqlx(type_name = "introspect.primary_def")] pub struct PgPrimary { diff --git a/crates/postgres/Cargo.toml b/crates/postgres/Cargo.toml index d48e2768..9624b2f6 100644 --- a/crates/postgres/Cargo.toml +++ b/crates/postgres/Cargo.toml @@ -26,3 +26,6 @@ starknet-types-core.workspace = true thiserror.workspace = true futures.workspace = true crc.workspace = true + + +torii-common.workspace = true diff --git a/crates/postgres/src/db.rs b/crates/postgres/src/db.rs index 13ba0e8d..ef5d608d 100644 --- a/crates/postgres/src/db.rs +++ b/crates/postgres/src/db.rs @@ -1,11 +1,9 @@ -use std::ops::Deref; - -pub use async_trait::async_trait; -use sqlx::Postgres; -use sqlx::{migrate::Migrator, Executor}; +use crate::migration::SchemaMigrator; +use async_trait::async_trait; +use sqlx::{migrate::Migrator, Postgres}; pub use sqlx::{PgPool, Transaction}; - -use crate::{migration::SchemaMigrator, SqlxResult}; +use std::ops::Deref; +use torii_common::sql::{Executable, SqlxResult}; #[async_trait] pub trait PostgresConnection { @@ -21,11 +19,9 @@ pub trait PostgresConnection { }; Ok(result?) } - async fn execute_queries(&self, queries: &[String]) -> SqlxResult<()> { + async fn execute_queries(&self, queries: impl Executable + Send) -> SqlxResult<()> { let mut transaction = self.begin().await?; - for query in queries { - transaction.execute(query.as_str()).await?; - } + queries.execute(&mut transaction).await?; transaction.commit().await } } diff --git a/crates/torii-common/Cargo.toml b/crates/torii-common/Cargo.toml index a10b65f1..9112e39f 100644 --- a/crates/torii-common/Cargo.toml +++ b/crates/torii-common/Cargo.toml @@ -15,6 +15,13 @@ urlencoding = "2" async-trait = "0.1" serde.workspace = true serde_json.workspace = true +sqlx = { workspace = true, features = [ + "postgres", + "runtime-tokio-rustls", + "mysql", + "sqlite", +] } +itertools.workspace = true [lints] workspace = true diff --git a/crates/torii-common/src/lib.rs b/crates/torii-common/src/lib.rs index 68eb7489..4bbb5573 100644 --- a/crates/torii-common/src/lib.rs +++ b/crates/torii-common/src/lib.rs @@ -5,6 +5,7 @@ pub mod json; pub mod metadata; +pub mod sql; pub mod token_uri; pub mod utils; diff --git a/crates/torii-common/src/sql.rs b/crates/torii-common/src/sql.rs new file mode 100644 index 00000000..531adaf0 --- /dev/null +++ b/crates/torii-common/src/sql.rs @@ -0,0 +1,241 @@ +use async_trait::async_trait; +use itertools::Itertools; +use sqlx::{Database, Executor, Postgres}; +pub use sqlx::{PgPool, Transaction}; +use std::borrow::Cow; +use std::sync::Arc; + +pub use sqlx::Error as SqlxError; + +pub type SqlxResult = std::result::Result; + +#[async_trait] +pub trait Executable { + async fn execute(self, transaction: &mut Transaction<'_, DB>) -> SqlxResult<()>; +} + +#[async_trait] +impl Executable for &str +where + for<'c> &'c mut ::Connection: Executor<'c, Database = DB>, +{ + async fn execute(self, transaction: &mut Transaction<'_, DB>) -> SqlxResult<()> { + transaction.execute(self).await?; + Ok(()) + } +} + +#[async_trait] +impl Executable for &String +where + for<'c> &'c mut ::Connection: Executor<'c, Database = DB>, +{ + async fn execute(self, transaction: &mut Transaction<'_, DB>) -> SqlxResult<()> { + transaction.execute(self.as_str()).await?; + Ok(()) + } +} + +#[async_trait] +impl Executable for String +where + for<'c> &'c mut ::Connection: Executor<'c, Database = DB>, +{ + async fn execute(self, transaction: &mut Transaction<'_, DB>) -> SqlxResult<()> { + transaction.execute(self.as_str()).await?; + Ok(()) + } +} + +#[async_trait] +impl Executable for Cow<'_, str> +where + for<'c> &'c mut ::Connection: Executor<'c, Database = DB>, +{ + async fn execute(self, transaction: &mut Transaction<'_, DB>) -> SqlxResult<()> { + transaction.execute(self.as_ref()).await?; + Ok(()) + } +} + +pub struct QueryLike, DB: Database> { + sql: S, + args: Option<::Arguments<'static>>, +} + +impl, DB: Database> QueryLike { + pub fn new(sql: impl Into, args: ::Arguments<'static>) -> Self { + Self { + sql: sql.into(), + args: Some(args), + } + } + + pub fn from_sql(sql: S) -> Self { + Self { sql, args: None } + } +} + +#[async_trait] +impl Executable for FlexStr +where + for<'c> &'c mut ::Connection: Executor<'c, Database = DB>, +{ + async fn execute(self, transaction: &mut Transaction<'_, DB>) -> SqlxResult<()> { + transaction.execute(self.as_ref()).await?; + Ok(()) + } +} + +pub enum FlexStr { + Owned(String), + Static(&'static str), + Shared(Arc), +} + +impl AsRef for FlexStr { + fn as_ref(&self) -> &str { + match self { + FlexStr::Owned(s) => s.as_str(), + FlexStr::Static(s) => s, + FlexStr::Shared(s) => s, + } + } +} + +impl PartialEq for FlexStr { + fn eq(&self, other: &str) -> bool { + self.as_ref() == other + } +} + +impl, DB: Database> PartialEq for QueryLike { + fn eq(&self, other: &str) -> bool { + self.sql.as_ref() == other + } +} + +impl From for FlexStr { + fn from(s: String) -> Self { + FlexStr::Owned(s) + } +} + +impl From<&'static str> for FlexStr { + fn from(s: &'static str) -> Self { + FlexStr::Static(s) + } +} + +impl From> for FlexStr { + fn from(s: Arc) -> Self { + FlexStr::Shared(s) + } +} + +impl From for QueryLike { + fn from(sql: FlexStr) -> Self { + QueryLike::from_sql(sql) + } +} + +impl From<&'static str> for QueryLike { + fn from(sql: &'static str) -> Self { + QueryLike::from_sql(sql.into()) + } +} + +impl From for QueryLike { + fn from(sql: String) -> Self { + QueryLike::from_sql(sql.into()) + } +} + +impl From> for QueryLike { + fn from(sql: Arc) -> Self { + QueryLike::from_sql(sql.into()) + } +} + +pub trait Queries { + fn add(&mut self, query: impl Into>); + fn adds(&mut self, queries: impl IntoIterator>>); +} + +impl Queries for Vec> { + fn add(&mut self, query: impl Into>) { + self.push(query.into()); + } + fn adds(&mut self, queries: impl IntoIterator>>) { + self.extend(queries.into_iter().map_into()); + } +} + +#[async_trait] +impl Executable for &[String; N] +where + for<'c> &'c mut ::Connection: Executor<'c, Database = DB>, +{ + async fn execute(self, transaction: &mut Transaction<'_, DB>) -> SqlxResult<()> { + for query in self { + transaction.execute(query.as_str()).await?; + } + Ok(()) + } +} + +#[async_trait] +impl + Send> Executable for Vec { + async fn execute(self, transaction: &mut Transaction<'_, DB>) -> SqlxResult<()> { + for item in self { + item.execute(transaction).await?; + } + Ok(()) + } +} + +#[async_trait] +impl<'a, DB: sqlx::Database, T> Executable for &'a Vec +where + &'a T: Executable + Send, + T: Send + Sync, +{ + async fn execute(self, transaction: &mut Transaction<'_, DB>) -> SqlxResult<()> { + for item in self { + item.execute(transaction).await?; + } + Ok(()) + } +} + +#[async_trait] +impl<'a, DB: sqlx::Database, T> Executable for &'a [T] +where + &'a T: Executable + Send, + T: Send + Sync, +{ + async fn execute(self, transaction: &mut Transaction<'_, DB>) -> SqlxResult<()> { + for item in self { + item.execute(transaction).await?; + } + Ok(()) + } +} + +#[async_trait] +impl + Send, DB: sqlx::Database> Executable for QueryLike +where + for<'c> &'c mut ::Connection: Executor<'c, Database = DB>, + for<'q> ::Arguments<'static>: sqlx::IntoArguments<'q, DB>, +{ + async fn execute(mut self, transaction: &mut Transaction<'_, DB>) -> SqlxResult<()> { + let sql = self.sql.as_ref(); + let args = self.args.take().unwrap_or_default(); + transaction.execute(sqlx::query_with(sql, args)).await?; + Ok(()) + } +} + +pub type PgQuery = QueryLike; +pub type MySqlQuery = QueryLike; +pub type SqliteQuery = QueryLike; diff --git a/examples/introspect/restart.rs b/examples/introspect/restart.rs index 4fff60c4..1221bd2d 100644 --- a/examples/introspect/restart.rs +++ b/examples/introspect/restart.rs @@ -8,11 +8,11 @@ use torii_introspect_postgres_sink::IntrospectPgDb; use torii_test_utils::{resolve_path_like, EventIterator, FakeProvider}; const DB_URL: &str = "postgres://torii:torii@localhost:5432/torii"; -const CHAIN_DATA_PATH: &str = "~/tc-tests/pistols-2"; -const SCHEMA_NAME: &str = "pistols"; -// const CHAIN_DATA_PATH: &str = "~/tc-tests/blob-arena"; -// const SCHEMA_NAME: &str = "blob_arena"; -const BATCH_SIZE: usize = 1000; +// const CHAIN_DATA_PATH: &str = "~/tc-tests/pistols-2"; +// const SCHEMA_NAME: &str = "pistols"; +const CHAIN_DATA_PATH: &str = "~/tc-tests/blob-arena-2"; +const SCHEMA_NAME: &str = "blob_arena"; +const BATCH_SIZE: usize = 10000; async fn run_events( events: &mut EventIterator, @@ -27,8 +27,6 @@ async fn run_events( decoder.store.initialize().await.unwrap(); decoder.load_tables(&[]).await.unwrap(); db.initialize_introspect_pg_sink().await.unwrap(); - db.load_tables_no_commit(decoder.get_tables().unwrap()) - .unwrap(); let mut running = true; let mut this_run = 0; while running { @@ -86,7 +84,7 @@ async fn main() { &mut event_iterator, provider.clone(), pool.clone(), - Some(50000), + Some(500000), &mut event_n, &mut success, ) From 30cc52ad0a237d44d0584886231e24e13f7e93e2 Mon Sep 17 00:00:00 2001 From: Bengineer Date: Thu, 19 Mar 2026 12:10:59 +0000 Subject: [PATCH 2/3] refactor: remove SqlxError and SqlxResult type aliases from Postgres module --- crates/postgres/src/lib.rs | 3 --- 1 file changed, 3 deletions(-) diff --git a/crates/postgres/src/lib.rs b/crates/postgres/src/lib.rs index e35d1ece..a2da4928 100644 --- a/crates/postgres/src/lib.rs +++ b/crates/postgres/src/lib.rs @@ -2,6 +2,3 @@ pub mod db; pub mod metadata; pub mod migration; pub use db::PostgresConnection; -pub use sqlx::Error as SqlxError; - -pub type SqlxResult = std::result::Result; From c4c99ca6ac15b05d99923ce5e7afca2847792b5c Mon Sep 17 00:00:00 2001 From: Bengineer Date: Thu, 19 Mar 2026 12:11:10 +0000 Subject: [PATCH 3/3] refactor: streamline SqlxResult usage and clean up imports across various modules --- Cargo.lock | 2 ++ crates/dojo/src/store/postgres.rs | 4 ++-- crates/dojo/src/store/sqlite.rs | 3 ++- crates/introspect-postgres-sink/src/query.rs | 3 +-- crates/introspect-sqlite-sink/src/processor.rs | 3 ++- crates/introspect/Cargo.toml | 2 ++ crates/introspect/src/postgres/global.rs | 3 ++- crates/introspect/src/postgres/mod.rs | 2 -- crates/introspect/src/postgres/owned.rs | 3 ++- crates/sqlite/Cargo.toml | 8 +++++++- crates/sqlite/src/db.rs | 2 +- crates/sqlite/src/lib.rs | 3 --- 12 files changed, 23 insertions(+), 15 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 820f99fb..e5cc1e94 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4850,6 +4850,7 @@ dependencies = [ "thiserror 2.0.18", "tonic-build", "torii", + "torii-common", ] [[package]] @@ -5036,6 +5037,7 @@ dependencies = [ "async-trait", "futures", "sqlx", + "torii-common", ] [[package]] diff --git a/crates/dojo/src/store/postgres.rs b/crates/dojo/src/store/postgres.rs index 2d22b00d..84229ec0 100644 --- a/crates/dojo/src/store/postgres.rs +++ b/crates/dojo/src/store/postgres.rs @@ -13,8 +13,9 @@ use sqlx::{FromRow, Postgres}; use starknet_types_core::felt::Felt; use std::collections::HashMap; use std::ops::Deref; +use torii_common::sql::SqlxResult; use torii_introspect::postgres::owned::PgTypeDef; -use torii_introspect::postgres::{PgFelt, SqlxResult}; +use torii_introspect::postgres::PgFelt; use torii_introspect::schema::ColumnKeyTrait; use torii_postgres::db::PostgresConnection; @@ -321,7 +322,6 @@ impl Deref for PgStore { impl PgStore { pub async fn initialize(&self) -> SqlxResult<()> { - println!("Running Dojo migrations..."); self.migrate(Some("dojo"), DOJO_STORE_MIGRATIONS).await } } diff --git a/crates/dojo/src/store/sqlite.rs b/crates/dojo/src/store/sqlite.rs index d4c05973..fd348d5e 100644 --- a/crates/dojo/src/store/sqlite.rs +++ b/crates/dojo/src/store/sqlite.rs @@ -12,9 +12,10 @@ use starknet_types_core::felt::Felt; use std::collections::HashMap; use std::io; use std::ops::Deref; +use torii_common::sql::SqlxResult; use torii_common::{blob_to_felt, felt_to_blob}; use torii_introspect::schema::ColumnKeyTrait; -use torii_sqlite::{SqliteConnection, SqlxResult}; +use torii_sqlite::SqliteConnection; pub const DOJO_SQLITE_STORE_MIGRATIONS: Migrator = sqlx::migrate!("./migrations/sqlite"); diff --git a/crates/introspect-postgres-sink/src/query.rs b/crates/introspect-postgres-sink/src/query.rs index 66e67e87..63b96983 100644 --- a/crates/introspect-postgres-sink/src/query.rs +++ b/crates/introspect-postgres-sink/src/query.rs @@ -6,10 +6,9 @@ use sqlx::Error::Encode as EncodeError; use sqlx::{postgres::PgArguments, types::Json}; use sqlx::{Arguments, Executor, Postgres}; use starknet_types_core::felt::Felt; -use torii_common::sql::{PgQuery, Queries}; +use torii_common::sql::{PgQuery, Queries, SqlxResult}; use torii_introspect::postgres::types::{PgPrimary, Uint128}; use torii_introspect::postgres::PgFelt; -use torii_postgres::SqlxResult; use crate::table::PgTable; use crate::{ diff --git a/crates/introspect-sqlite-sink/src/processor.rs b/crates/introspect-sqlite-sink/src/processor.rs index 3f6121b2..4eaaf92a 100644 --- a/crates/introspect-sqlite-sink/src/processor.rs +++ b/crates/introspect-sqlite-sink/src/processor.rs @@ -2,6 +2,7 @@ use crate::json::SqliteJsonSerializer; use crate::table::{SqliteTable, SqliteTableError}; use crate::INTROSPECT_SQLITE_SINK_MIGRATIONS; use serde_json::{Serializer as JsonSerializer, Value}; +use sqlx::Error as SqlxError; use sqlx::Row; use starknet_types_core::felt::Felt; use std::collections::HashMap; @@ -12,7 +13,7 @@ use torii::etl::envelope::MetaData; use torii_introspect::events::{IntrospectBody, IntrospectMsg}; use torii_introspect::schema::TableSchema; use torii_introspect::InsertsFields; -use torii_sqlite::{SqliteConnection, SqlxError}; +use torii_sqlite::SqliteConnection; #[derive(Debug, thiserror::Error)] pub enum SqliteDbError { diff --git a/crates/introspect/Cargo.toml b/crates/introspect/Cargo.toml index 4021f5d0..4e00c48e 100644 --- a/crates/introspect/Cargo.toml +++ b/crates/introspect/Cargo.toml @@ -27,6 +27,8 @@ sqlx = { workspace = true, features = [ async-trait.workspace = true bigdecimal.workspace = true +torii-common.workspace = true + [build-dependencies] tonic-build = "0.12" diff --git a/crates/introspect/src/postgres/global.rs b/crates/introspect/src/postgres/global.rs index 10123292..41f23449 100644 --- a/crates/introspect/src/postgres/global.rs +++ b/crates/introspect/src/postgres/global.rs @@ -6,8 +6,9 @@ use itertools::Itertools; use sqlx::types::Json; use sqlx::{FromRow, PgPool}; use starknet_types_core::felt::Felt; +use torii_common::sql::SqlxResult; -use crate::postgres::{attribute_type, felt252_type, string_type, PgAttribute, PgFelt, SqlxResult}; +use crate::postgres::{attribute_type, felt252_type, string_type, PgAttribute, PgFelt}; #[derive(FromRow)] pub struct ColumnRow { diff --git a/crates/introspect/src/postgres/mod.rs b/crates/introspect/src/postgres/mod.rs index 63df01f5..ccb5c023 100644 --- a/crates/introspect/src/postgres/mod.rs +++ b/crates/introspect/src/postgres/mod.rs @@ -3,5 +3,3 @@ pub mod owned; pub mod store; pub mod types; pub use types::{attribute_type, felt252_type, string_type, PgAttribute, PgFelt}; - -pub type SqlxResult = Result; diff --git a/crates/introspect/src/postgres/owned.rs b/crates/introspect/src/postgres/owned.rs index d88bd569..c781c520 100644 --- a/crates/introspect/src/postgres/owned.rs +++ b/crates/introspect/src/postgres/owned.rs @@ -1,5 +1,5 @@ use crate::postgres::types::PgPrimary; -use crate::postgres::{PgAttribute, PgFelt, SqlxResult}; +use crate::postgres::{PgAttribute, PgFelt}; use crate::schema::{ColumnKeyTrait, Table}; use async_trait::async_trait; use introspect_types::{Attribute, ColumnDef, ColumnInfo, PrimaryDef, TypeDef}; @@ -10,6 +10,7 @@ use sqlx::types::Json; use sqlx::{FromRow, PgPool, Postgres}; use starknet_types_core::felt::Felt; use std::collections::HashMap; +use torii_common::sql::SqlxResult; pub const TABLE_INSERT_QUERY: &str = " INSERT INTO introspect.tables (owner, id, name, attributes, primary_def, column_ids, updated_at, created_block, updated_block, created_tx, updated_tx) diff --git a/crates/sqlite/Cargo.toml b/crates/sqlite/Cargo.toml index 44b39e24..08f62825 100644 --- a/crates/sqlite/Cargo.toml +++ b/crates/sqlite/Cargo.toml @@ -7,7 +7,13 @@ description = "SQLite connection helpers for Torii storage crates" [dependencies] async-trait.workspace = true futures.workspace = true -sqlx = { workspace = true, features = ["sqlite", "runtime-tokio-rustls", "migrate"] } +sqlx = { workspace = true, features = [ + "sqlite", + "runtime-tokio-rustls", + "migrate", +] } + +torii-common.workspace = true [lints] workspace = true diff --git a/crates/sqlite/src/db.rs b/crates/sqlite/src/db.rs index 7ad4d931..c9f4a200 100644 --- a/crates/sqlite/src/db.rs +++ b/crates/sqlite/src/db.rs @@ -6,9 +6,9 @@ use sqlx::migrate::Migrator; use sqlx::sqlite::SqliteConnectOptions; use sqlx::Sqlite; pub use sqlx::{SqlitePool, Transaction}; +use torii_common::sql::SqlxResult; use crate::migration::NamespaceMigrator; -use crate::SqlxResult; #[async_trait] pub trait SqliteConnection { diff --git a/crates/sqlite/src/lib.rs b/crates/sqlite/src/lib.rs index 75ed50fe..e0e1dc65 100644 --- a/crates/sqlite/src/lib.rs +++ b/crates/sqlite/src/lib.rs @@ -2,6 +2,3 @@ pub mod db; pub mod migration; pub use db::{is_sqlite_memory_path, sqlite_connect_options, SqliteConnection}; -pub use sqlx::Error as SqlxError; - -pub type SqlxResult = std::result::Result;