feat(cu): support using postgres as db #588

servers/cu/.postgres/Dockerfile

@@ -0,0 +1,6 @@
+# Only used for development purposes
+FROM postgres:17
+
+ENV POSTGRES_PASSWORD=admin
+ENV POSTGRES_USER=admin
+ENV POSTGRES_DB=cu

servers/cu/package.json

@@ -33,6 +33,7 @@
     "opossum": "^8.4.0",
     "p-map": "^7.0.3",
     "p-queue": "^8.0.1",
+    "pg": "^8.13.1",
     "prom-client": "^15.1.3",
     "ramda": "^0.30.1",
     "undici": "^7.2.0",

servers/cu/src/bootstrap.js

@@ -14,8 +14,8 @@ import { fromPromise } from 'hyper-async'
 import lt from 'long-timeout'
 
 // Precanned clients to use for OOTB apis
+import * as DbClient from './effects/db.js'
 import * as ArweaveClient from './effects/arweave.js'
-import * as SqliteClient from './effects/sqlite.js'
 import * as AoSuClient from './effects/ao-su.js'
 import * as WasmClient from './effects/wasm.js'
 import * as AoProcessClient from './effects/ao-process.js'
@@ -77,8 +77,7 @@ export const createApis = async (ctx) => {
     cacheKeyFn: ({ processId }) => processId
   })
 
-  const DB_URL = `${ctx.DB_URL}.sqlite`
-  const sqlite = await SqliteClient.createSqliteClient({ url: DB_URL, bootstrap: true })
+  const db = await DbClient.createDbClient({ url: ctx.DB_URL, bootstrap: true })
 
   const BROADCAST = 'workers'
   const workerBroadcast = new BroadcastChannel(BROADCAST).unref()
@@ -98,7 +97,7 @@ export const createApis = async (ctx) => {
           ARWEAVE_URL: ctx.ARWEAVE_URL,
           GRAPHQL_URL: ctx.GRAPHQL_URL,
           CHECKPOINT_GRAPHQL_URL: ctx.CHECKPOINT_GRAPHQL_URL,
-          DB_URL,
+          DB_URL: ctx.DB_URL,
           id: workerId,
           MODE: ctx.MODE,
           LOG_CONFIG_PATH: ctx.LOG_CONFIG_PATH,
@@ -219,9 +218,9 @@ export const createApis = async (ctx) => {
     hashWasmMemory: WasmClient.hashWasmMemoryWith({ logger: ctx.logger }),
     buildAndSignDataItem: ArweaveClient.buildAndSignDataItemWith({ WALLET: ctx.WALLET }),
     uploadDataItem: ArweaveClient.uploadDataItemWith({ UPLOADER_URL: ctx.UPLOADER_URL, fetch: ctx.fetch, logger: ctx.logger }),
-    writeCheckpointRecord: AoProcessClient.writeCheckpointRecordWith({ db: sqlite }),
+    writeCheckpointRecord: AoProcessClient.writeCheckpointRecordWith({ db }),
     writeFileCheckpointMemory,
-    writeFileCheckpointRecord: AoProcessClient.writeFileCheckpointRecordWith({ db: sqlite }),
+    writeFileCheckpointRecord: AoProcessClient.writeFileCheckpointRecordWith({ db }),
     logger: ctx.logger,
     DISABLE_PROCESS_CHECKPOINT_CREATION: ctx.DISABLE_PROCESS_CHECKPOINT_CREATION,
     DISABLE_PROCESS_FILE_CHECKPOINT_CREATION: ctx.DISABLE_PROCESS_FILE_CHECKPOINT_CREATION,
@@ -307,14 +306,14 @@ export const createApis = async (ctx) => {
     loadTransactionMeta: ArweaveClient.loadTransactionMetaWith({ fetch: ctx.fetch, GRAPHQL_URL: ctx.GRAPHQL_URL, logger }),
     loadTransactionData: ArweaveClient.loadTransactionDataWith({ fetch: ctx.fetch, ARWEAVE_URL: ctx.ARWEAVE_URL, logger }),
     isProcessOwnerSupported: AoProcessClient.isProcessOwnerSupportedWith({ ALLOW_OWNERS: ctx.ALLOW_OWNERS }),
-    findProcess: AoProcessClient.findProcessWith({ db: sqlite, logger }),
+    findProcess: AoProcessClient.findProcessWith({ db, logger }),
     findLatestProcessMemory: AoProcessClient.findLatestProcessMemoryWith({
       cache: wasmMemoryCache,
       loadTransactionData: ArweaveClient.loadTransactionDataWith({ fetch: ctx.fetch, ARWEAVE_URL: ctx.ARWEAVE_URL, logger }),
       readProcessMemoryFile,
       readFileCheckpointMemory,
-      findFileCheckpointBefore: AoProcessClient.findFileCheckpointBeforeWith({ db: sqlite }),
-      findRecordCheckpointBefore: AoProcessClient.findRecordCheckpointBeforeWith({ db: sqlite }),
+      findFileCheckpointBefore: AoProcessClient.findFileCheckpointBeforeWith({ db }),
+      findRecordCheckpointBefore: AoProcessClient.findRecordCheckpointBeforeWith({ db }),
       address,
       queryGateway: ArweaveClient.queryGatewayWith({ fetch: ctx.fetch, GRAPHQL_URL: ctx.GRAPHQL_URL, logger }),
       queryCheckpointGateway: ArweaveClient.queryGatewayWith({ fetch: ctx.fetch, GRAPHQL_URL: ctx.CHECKPOINT_GRAPHQL_URL, logger }),
@@ -331,14 +330,14 @@ export const createApis = async (ctx) => {
     }),
     evaluationCounter,
     // gasCounter,
-    saveProcess: AoProcessClient.saveProcessWith({ db: sqlite, logger }),
-    findEvaluation: AoEvaluationClient.findEvaluationWith({ db: sqlite, logger }),
-    saveEvaluation: AoEvaluationClient.saveEvaluationWith({ db: sqlite, logger }),
-    findBlocks: AoBlockClient.findBlocksWith({ db: sqlite, logger }),
-    saveBlocks: AoBlockClient.saveBlocksWith({ db: sqlite, logger }),
+    saveProcess: AoProcessClient.saveProcessWith({ db, logger }),
+    findEvaluation: AoEvaluationClient.findEvaluationWith({ db, logger }),
+    saveEvaluation: AoEvaluationClient.saveEvaluationWith({ db, logger }),
+    findBlocks: AoBlockClient.findBlocksWith({ db, logger }),
+    saveBlocks: AoBlockClient.saveBlocksWith({ db, logger }),
     loadBlocksMeta: AoBlockClient.loadBlocksMetaWith({ fetch: ctx.fetch, GRAPHQL_URL: ctx.GRAPHQL_URL, pageSize: 90, logger }),
-    findModule: AoModuleClient.findModuleWith({ db: sqlite, logger }),
-    saveModule: AoModuleClient.saveModuleWith({ db: sqlite, logger }),
+    findModule: AoModuleClient.findModuleWith({ db, logger }),
+    saveModule: AoModuleClient.saveModuleWith({ db, logger }),
     loadEvaluator: AoModuleClient.evaluatorWith({
       loadWasmModule,
       evaluateWith: (prep) => primaryWorkQueue.add(() =>
@@ -362,7 +361,7 @@ export const createApis = async (ctx) => {
       ),
       logger
     }),
-    findMessageBefore: AoEvaluationClient.findMessageBeforeWith({ db: sqlite, logger }),
+    findMessageBefore: AoEvaluationClient.findMessageBeforeWith({ db, logger }),
     loadTimestamp: AoSuClient.loadTimestampWith({ fetch: ctx.fetch, logger }),
     loadProcess: AoSuClient.loadProcessWith({ fetch: ctx.fetch, logger }),
     loadMessages: AoSuClient.loadMessagesWith({ fetch: ctx.fetch, pageSize: 1000, logger }),
@@ -450,13 +449,13 @@ export const createApis = async (ctx) => {
   const readCronResultsLogger = ctx.logger.child('readCronResults')
   const readCronResults = readCronResultsWith({
     ...sharedDeps(readCronResultsLogger),
-    findEvaluations: AoEvaluationClient.findEvaluationsWith({ db: sqlite, logger: readCronResultsLogger })
+    findEvaluations: AoEvaluationClient.findEvaluationsWith({ db, logger: readCronResultsLogger })
   })
 
   const readResultsLogger = ctx.logger.child('readResults')
   const readResults = readResultsWith({
     ...sharedDeps(readResultsLogger),
-    findEvaluations: AoEvaluationClient.findEvaluationsWith({ db: sqlite, logger: readResultsLogger })
+    findEvaluations: AoEvaluationClient.findEvaluationsWith({ db, logger: readResultsLogger })
   })
 
   let checkpointP

servers/cu/src/domain/model.js

@@ -411,7 +411,15 @@ export const evaluationSchema = z.object({
   evaluatedAt: z.preprocess(
     (
       arg
-    ) => (typeof arg === 'string' || arg instanceof Date ? new Date(arg) : arg),
+    ) => {
+      // typeof arg === 'string' || arg instanceof Date ? new Date(arg + 0) : arg
+
+      if (arg instanceof Date) return arg
+      if (typeof arg === 'string') try { arg = parseInt(arg) } catch {}
+      if (typeof arg === 'number') return new Date(arg)
+
+      return arg
+    },
     z.date()
   ),
   /**

servers/cu/src/effects/ao-block.js

@@ -2,10 +2,11 @@ import { fromPromise, of } from 'hyper-async'
 import { applySpec, last, map, path, pathSatisfies, pipe, pluck, prop, props, splitEvery } from 'ramda'
 import { z } from 'zod'
 import pMap from 'p-map'
+import CircuitBreaker from 'opossum'
+
 import { blockSchema } from '../domain/model.js'
-import { BLOCKS_TABLE } from './sqlite.js'
 import { backoff, strFromFetchError } from '../domain/utils.js'
-import CircuitBreaker from 'opossum'
+import { BLOCKS_TABLE } from './db.js'
 
 const okRes = (res) => {
   if (res.ok) return res

servers/cu/src/effects/ao-evaluation.js

@@ -3,7 +3,7 @@ import { always, applySpec, isEmpty, isNotNil, converge, mergeAll, map, unapply,
 import { z } from 'zod'
 
 import { evaluationSchema } from '../domain/model.js'
-import { EVALUATIONS_TABLE, MESSAGES_TABLE, COLLATION_SEQUENCE_MAX_CHAR } from './sqlite.js'
+import { EVALUATIONS_TABLE, MESSAGES_TABLE, COLLATION_SEQUENCE_MAX_CHAR } from './db.js'
 
 const evaluationDocSchema = z.object({
   id: z.string().min(1),
@@ -60,8 +60,7 @@ const toEvaluation = applySpec({
 
 const fromEvaluationDoc = pipe(
   evolve({
-    output: JSON.parse,
-    evaluatedAt: (timestamp) => new Date(timestamp)
+    output: JSON.parse
   }),
   /**
    * Ensure the input matches the expected
@@ -76,8 +75,8 @@ export function findEvaluationWith ({ db }) {
     return {
       sql: `
         SELECT
-          id, processId, messageId, deepHash, nonce, epoch, timestamp,
-          ordinate, blockHeight, cron, evaluatedAt, output
+          id, "processId", "messageId", "deepHash", nonce, epoch, timestamp,
+          ordinate, "blockHeight", cron, "evaluatedAt", output
         FROM ${EVALUATIONS_TABLE}
         WHERE
           id = ?;
@@ -143,7 +142,7 @@ export function saveEvaluationWith ({ db, logger: _logger }) {
       {
         sql: `
           INSERT OR IGNORE INTO ${EVALUATIONS_TABLE}
-            (id, processId, messageId, deepHash, nonce, epoch, timestamp, ordinate, blockHeight, cron, evaluatedAt, output)
+            (id, "processId", "messageId", "deepHash", nonce, epoch, timestamp, ordinate, "blockHeight", cron, "evaluatedAt", output)
             VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?);
         `,
         parameters: [
@@ -170,7 +169,7 @@ export function saveEvaluationWith ({ db, logger: _logger }) {
     if (!evaluation.cron) {
       statements.push({
         sql: `
-          INSERT OR IGNORE INTO ${MESSAGES_TABLE} (id, processId, seq) VALUES (?, ?, ?);
+          INSERT OR IGNORE INTO ${MESSAGES_TABLE} (id, "processId", seq) VALUES (?, ?, ?);
          `,
         parameters: messageDocParamsSchema.parse([
           createMessageId({
@@ -204,8 +203,8 @@ export function findEvaluationsWith ({ db }) {
     return {
       sql: `
         SELECT
-          id, processId, messageId, deepHash, nonce, epoch, timestamp,
-          ordinate, blockHeight, cron, evaluatedAt, output
+          id, "processId", "messageId", "deepHash", nonce, epoch, timestamp,
+          ordinate, "blockHeight", cron, "evaluatedAt", output
         FROM ${EVALUATIONS_TABLE}
         WHERE
           id > ? AND id <= ?
@@ -251,7 +250,7 @@ export function findMessageBeforeWith ({ db }) {
         FROM ${MESSAGES_TABLE}
         WHERE
           id = ?
-          AND processId = ?
+          AND "processId" = ?
           AND seq < ?
         LIMIT 1;
       `,

servers/cu/src/effects/ao-evaluation.test.js

@@ -5,7 +5,7 @@ import assert from 'node:assert'
 import { createTestLogger } from '../domain/logger.js'
 import { findEvaluationSchema, findEvaluationsSchema, findMessageBeforeSchema, saveEvaluationSchema } from '../domain/dal.js'
 import { findMessageBeforeWith, findEvaluationWith, findEvaluationsWith, saveEvaluationWith } from './ao-evaluation.js'
-import { COLLATION_SEQUENCE_MAX_CHAR, EVALUATIONS_TABLE, MESSAGES_TABLE } from './sqlite.js'
+import { COLLATION_SEQUENCE_MAX_CHAR, EVALUATIONS_TABLE, MESSAGES_TABLE } from './db.js'
 
 const logger = createTestLogger({ name: 'ao-cu:readState' })