ClickHouse mirror

codegenerator/cli/npm/envio/package.json

@@ -20,6 +20,7 @@
   },
   "homepage": "https://envio.dev",
   "dependencies": {
+    "@clickhouse/client": "1.12.1",
     "@envio-dev/hypersync-client": "0.6.6",
     "@envio-dev/hyperfuel-client": "1.2.2",
     "rescript": "11.1.3",

codegenerator/cli/npm/envio/package.json.tmpl

@@ -31,6 +31,7 @@
     "envio-darwin-arm64": "${version}"
   },
   "dependencies": {
+    "@clickhouse/client": "1.12.1",
     "@envio-dev/hypersync-client": "0.6.6",
     "@envio-dev/hyperfuel-client": "1.2.2",
     "rescript": "11.1.3",

codegenerator/cli/npm/envio/src/Internal.res

@@ -319,8 +319,8 @@ let makeCacheTable = (~effectName) => {
   Table.mkTable(
     cacheTablePrefix ++ effectName,
     ~fields=[
-      Table.mkField("id", Text, ~fieldSchema=S.string, ~isPrimaryKey=true),
-      Table.mkField("output", JsonB, ~fieldSchema=cacheOutputSchema, ~isNullable=true),
+      Table.mkField("id", String, ~fieldSchema=S.string, ~isPrimaryKey=true),
+      Table.mkField("output", Json, ~fieldSchema=cacheOutputSchema, ~isNullable=true),
     ],
   )
 }

codegenerator/cli/npm/envio/src/Mirror.res

@@ -0,0 +1,29 @@
+type t = {
+  initialize: (
+    ~chainConfigs: array<Config.chain>=?,
+    ~entities: array<Internal.entityConfig>=?,
+    ~enums: array<Internal.enumConfig<Internal.enum>>=?,
+  ) => promise<unit>,
+  setOrThrow: 'item. (
+    ~items: array<'item>,
+    ~table: Table.table,
+    ~itemSchema: S.t<'item>,
+  ) => promise<unit>,
+}
+
+let makeClickHouse = (~host, ~database, ~username, ~password): t => {
+  let client = ClickHouse.createClient({
+    url: host,
+    username,
+    password,
+  })
+
+  {
+    initialize: (~chainConfigs as _=[], ~entities=[], ~enums=[]) => {
+      ClickHouse.initialize(client, ~database, ~entities, ~enums)
+    },
+    setOrThrow: (~items, ~table, ~itemSchema) => {
+      ClickHouse.setOrThrow(client, ~items, ~table, ~itemSchema, ~database)
+    },
+  }
+}

codegenerator/cli/npm/envio/src/PgStorage.res

@@ -32,12 +32,13 @@ let makeCreateTableQuery = (table: Table.table, ~pgSchema, ~isNumericArrayAsText
       let fieldName = field->Table.getDbFieldName
 
       {
-        `"${fieldName}" ${switch fieldType {
-          | Custom(name) if !(name->Js.String2.startsWith("NUMERIC(")) => `"${pgSchema}".${name}`
-          // Workaround for Hasura bug https://github.com/enviodev/hyperindex/issues/788
-          | Numeric if isArray && isNumericArrayAsText => (Table.Text :> string)
-          | _ => (fieldType :> string)
-          }}${isArray ? "[]" : ""}${switch defaultValue {
+        `"${fieldName}" ${Table.getPgFieldType(
+            ~fieldType,
+            ~pgSchema,
+            ~isArray,
+            ~isNullable,
+            ~isNumericArrayAsText,
+          )}${switch defaultValue {
           | Some(defaultValue) => ` DEFAULT ${defaultValue}`
           | None => isNullable ? `` : ` NOT NULL`
           }}`
@@ -561,6 +562,7 @@ let rec writeBatch = async (
   ~allEntities,
   ~setEffectCacheOrThrow,
   ~batchCache,
+  ~mirror: option<Mirror.t>=?,
   ~escapeTables=?,
 ) => {
   try {
@@ -685,6 +687,21 @@ let rec writeBatch = async (
                 ),
               )
               ->ignore
+
+              // Mirror to ClickHouse if configured
+              switch mirror {
+              | Some(mirror) =>
+                promises
+                ->Js.Array2.push(
+                  mirror.setOrThrow(
+                    ~items=batchSetUpdates,
+                    ~itemSchema=entityConfig.entityHistory.setUpdateSchema,
+                    ~table=entityConfig.entityHistory.table,
+                  ),
+                )
+                ->ignore
+              | None => ()
+              }
             }
           }
 
@@ -863,6 +880,7 @@ let rec writeBatch = async (
       ~setEffectCacheOrThrow,
       ~batchCache,
       ~allEntities,
+      ~mirror?,
     )
   }
 }
@@ -876,6 +894,7 @@ let make = (
   ~pgDatabase,
   ~pgPassword,
   ~isHasuraEnabled,
+  ~mirror: option<Mirror.t>=?,
   ~onInitialize=?,
   ~onNewTables=?,
 ): Persistence.storage => {
@@ -1007,6 +1026,12 @@ let make = (
       )
     }
 
+    // Call mirror.initialize before executing PG queries
+    switch mirror {
+    | Some(mirror) => await mirror.initialize(~chainConfigs, ~entities, ~enums)
+    | None => ()
+    }
+
     let queries = makeInitializeTransaction(
       ~pgSchema,
       ~pgUser,
@@ -1367,6 +1392,7 @@ let make = (
       ~allEntities,
       ~setEffectCacheOrThrow,
       ~batchCache,
+      ~mirror?,
     )
   }
 

codegenerator/cli/npm/envio/src/Prometheus.res

@@ -16,6 +16,18 @@ let executeBatchDurationCounter = PromClient.Counter.makeCounter({
   "labelNames": [],
 })
 
+let storageWriteTimeCounter = PromClient.Counter.makeCounter({
+  "name": "envio_storage_write_time",
+  "help": "Cumulative time spent writing batches to storage in milliseconds",
+  "labelNames": [],
+})
+
+let storageWriteCounter = PromClient.Counter.makeCounter({
+  "name": "envio_storage_write_count",
+  "help": "Total number of batch writes to storage",
+  "labelNames": [],
+})
+
 let allChainsSyncedToHead = PromClient.Gauge.makeGauge({
   "name": "hyperindex_synced_to_head",
   "help": "All chains fully synced",
@@ -213,6 +225,14 @@ let incrementExecuteBatchDurationCounter = (~duration) => {
   executeBatchDurationCounter->PromClient.Counter.incMany(duration)
 }
 
+let incrementStorageWriteTimeCounter = (~duration) => {
+  storageWriteTimeCounter->PromClient.Counter.incMany(duration)
+}
+
+let incrementStorageWriteCounter = () => {
+  storageWriteCounter->PromClient.Counter.inc
+}
+
 let setSourceChainHeight = (~blockNumber, ~chainId) => {
   sourceChainHeight
   ->PromClient.Gauge.labels({"chainId": chainId})

codegenerator/cli/npm/envio/src/bindings/ClickHouse.res

@@ -0,0 +1,169 @@
+// ClickHouse client bindings for @clickhouse/client
+
+type client
+
+type clientConfig = {
+  url?: string,
+  database?: string,
+  username?: string,
+  password?: string,
+}
+
+type execParams = {query: string}
+
+@module("@clickhouse/client")
+external createClient: clientConfig => client = "createClient"
+
+@send
+external exec: (client, execParams) => promise<unit> = "exec"
+
+@send
+external close: client => promise<unit> = "close"
+
+type insertParams<'a> = {
+  table: string,
+  values: array<'a>,
+  format: string,
+}
+
+@send
+external insert: (client, insertParams<'a>) => promise<unit> = "insert"
+
+let getClickHouseFieldType = (
+  ~fieldType: Table.fieldType,
+  ~isNullable: bool,
+  ~isArray: bool,
+): string => {
+  let baseType = switch fieldType {
+  | Int32 => "Int32"
+  | Uint32 => "UInt32"
+  | Serial => "Int32"
+  | BigInt({?precision}) =>
+    switch precision {
+    | None => "String" // Fallback for unbounded BigInt
+    | Some(precision) =>
+      if precision > 38 {
+        "String"
+      } else {
+        `Decimal(${precision->Js.Int.toString},0)`
+      }
+    }
+  | BigDecimal({?config}) =>
+    switch config {
+    | None => "String" // Fallback for unbounded BigInt
+    | Some((precision, scale)) =>
+      if precision > 38 || scale > precision {
+        "String"
+      } else {
+        `Decimal(${precision->Js.Int.toString},${scale->Js.Int.toString})`
+      }
+    }
+  | Boolean => "Bool"
+  | Number => "Float64"
+  | String => "String"
+  | Json => "String"
+  | Date => "DateTime64(3, 'UTC')"
+  | Enum(_) => "String"
+  | Entity(_) => "String"
+  }
+
+  let baseType = if isArray {
+    `Array(${baseType})`
+  } else {
+    baseType
+  }
+
+  isNullable ? `Nullable(${baseType})` : baseType
+}
+
+let setOrThrow = async (
+  client,
+  ~items: array<'item>,
+  ~table: Table.table,
+  ~itemSchema: S.t<'item>,
+  ~database: string,
+) => {
+  if items->Array.length === 0 {
+    ()
+  } else {
+    try {
+      // Convert entity updates to ClickHouse row format
+      let values = items->Js.Array2.map(item => {
+        item->S.reverseConvertOrThrow(itemSchema)
+      })
+
+      await client->insert({
+        table: `${database}.\`${table.tableName}\``,
+        values,
+        format: "JSONEachRow",
+      })
+    } catch {
+    | exn =>
+      raise(
+        Persistence.StorageError({
+          message: `Failed to insert items into ClickHouse table "${table.tableName}"`,
+          reason: exn->Utils.prettifyExn,
+        }),
+      )
+    }
+  }
+}
+
+// Generate CREATE TABLE query for entity history table
+let makeCreateHistoryTableQuery = (entity: Internal.entityConfig, ~database: string) => {
+  let historyTable = entity.entityHistory.table
+
+  let fieldDefinitions =
+    historyTable.fields
+    ->Belt.Array.keepMap(field => {
+      switch field {
+      | Field(field) =>
+        Some({
+          let fieldName = field->Table.getDbFieldName
+          let clickHouseType = getClickHouseFieldType(
+            ~fieldType=field.fieldType,
+            ~isNullable=field.isNullable,
+            ~isArray=field.isArray,
+          )
+          `\`${fieldName}\` ${clickHouseType}`
+        })
+      | DerivedFrom(_) => None
+      }
+    })
+    ->Js.Array2.joinWith(",\n  ")
+
+  let tableName = historyTable.tableName
+
+  `CREATE TABLE IF NOT EXISTS ${database}.\`${tableName}\` (
+  ${fieldDefinitions}
+)
+ENGINE = MergeTree()
+ORDER BY (id, ${EntityHistory.checkpointIdFieldName})`
+}
+
+// Initialize ClickHouse tables for entities
+let initialize = async (
+  client,
+  ~database: string,
+  ~entities: array<Internal.entityConfig>,
+  ~enums as _: array<Internal.enumConfig<Internal.enum>>,
+) => {
+  try {
+    await client->exec({query: `DROP DATABASE IF EXISTS ${database}`})
+    await client->exec({query: `CREATE DATABASE ${database}`})
+    await client->exec({query: `USE ${database}`})
+
+    await Promise.all(
+      entities->Belt.Array.map(entity =>
+        client->exec({query: makeCreateHistoryTableQuery(entity, ~database)})
+      ),
+    )->Promise.ignoreValue
+
+    Logging.trace("ClickHouse mirror initialization completed successfully")
+  } catch {
+  | exn => {
+      Logging.errorWithExn(exn, "Failed to initialize ClickHouse mirror")
+      Js.Exn.raiseError("ClickHouse initialization failed")
+    }
+  }
+}