feat(duckdb): ✨ Add duckdb-async support and enhance DuckDB tools

Author: flowcore-platform

bun.lock

@@ -17,6 +17,7 @@
         "aws-sdk": "^2.1692.0",
         "dayjs": "^1.11.13",
         "duckdb": "^1.2.0",
+        "duckdb-async": "^1.2.0",
         "jwt-decode": "^4.0.0",
         "mock-aws-s3": "^4.0.2",
         "nock": "^14.0.1",
@@ -209,6 +210,8 @@
 
     "duckdb": ["[email protected]", "", { "dependencies": { "@mapbox/node-pre-gyp": "^2.0.0", "node-addon-api": "^7.0.0", "node-gyp": "^9.3.0" } }, "sha512-zAHHRTMoZhWIwvOsyNkgV9c1nq0gR0j+ZyX0uTCRFZTNOlYO4lnErP5Fddt/6iKMXsTNL9v1oTG9E76S5jMh7w=="],
 
+    "duckdb-async": ["[email protected]", "", { "dependencies": { "duckdb": "^1.2.0" } }, "sha512-Ri8PL/TGZRhbW+53a/zqD4nJdH3p+mRB6owb7Hl5tXu3e1rvXVLil543aiVEnVKKxSHSBUwHyoutt+rOHrhc6g=="],
+
     "dunder-proto": ["[email protected]", "", { "dependencies": { "call-bind-apply-helpers": "^1.0.1", "es-errors": "^1.3.0", "gopd": "^1.2.0" } }, "sha512-KIN/nDJBQRcXw0MLVhZE9iQHmG68qAVIBg9CqmUYjmQIhgij9U5MFvrqkUL5FbtyyzZuOeOt0zdeRe4UY7ct+A=="],
 
     "eastasianwidth": ["[email protected]", "", {}, "sha512-I88TYZWc9XiYHRQ4/3c5rjjfgkjhLyW2luGIheGERbNQ6OY7yTybanSpDXZa8y7VUP9YmDcYa+eyq4ca7iLqWA=="],

package.json

@@ -44,6 +44,7 @@
     "aws-sdk": "^2.1692.0",
     "dayjs": "^1.11.13",
     "duckdb": "^1.2.0",
+    "duckdb-async": "^1.2.0",
     "jwt-decode": "^4.0.0",
     "mock-aws-s3": "^4.0.2",
     "nock": "^14.0.1",

src/duckdb/index.ts

@@ -1,44 +1,50 @@
-import { type DuckDBConnection, DuckDBInstance } from "@duckdb/node-api"
+import { type Connection, Database } from "duckdb-async"
 
-let instance: DuckDBInstance | null = null
-let connection: DuckDBConnection | null = null
+let db: Database | null = null
+let connection: Connection | null = null
 
 /**
  * Initializes an in-memory DuckDB database
  * @returns A reference to the initialized database
  */
-export async function initializeDuckDB(databaseFile?: string) {
-  if (instance !== null) {
-    return { success: true, databaseInstance: instance }
-  }
-
+export async function initializeDuckDB(
+  file?: string,
+): Promise<{ success: boolean; databaseInstance?: Database; error?: string }> {
   try {
-    // Create a new DuckDB instance with a persistent database file
-    instance = await DuckDBInstance.create(databaseFile || ":memory:")
-    connection = await instance.connect()
+    // Create a new DuckDB instance with a persistent database file or in-memory
+    db = await Database.create(file || ":memory:")
+    connection = await db.connect()
+
+    // Verify connection is working
+    await connection.exec("SELECT 1")
 
-    return { success: true, databaseInstance: instance }
+    return { success: true, databaseInstance: db }
   } catch (error) {
     return { success: false, error: `Failed to initialize DuckDB: ${error}` }
   }
 }
 
+export async function closeDuckDB() {
+  if (db) {
+    await db.close()
+  }
+}
+
 /**
  * Creates a table in the DuckDB database using raw SQL
  * @param createTableSQL The SQL CREATE TABLE statement
  * @returns Object indicating success or failure
  */
-export async function createTableWithSQL(createTableSQL: string) {
-  if (!instance || !connection) {
+export async function createTableWithSQL(createTableSQL: string): Promise<{ success: boolean; message: string }> {
+  if (!db || !connection) {
     return {
       success: false,
       message: "Database not initialized. Call initializeDuckDB first.",
     }
   }
 
   try {
-    await connection.run(createTableSQL)
-
+    await connection.exec(createTableSQL)
     return {
       success: true,
       message: "Table created successfully",
@@ -61,8 +67,8 @@ export async function createTableWithSQL(createTableSQL: string) {
 export async function createProjectionTable(
   tableName: string,
   columns: Array<{ name: string; type: string; constraints?: string }>,
-) {
-  if (!instance || !connection) {
+): Promise<{ success: boolean; message: string }> {
+  if (!db || !connection) {
     return {
       success: false,
       message: "Database not initialized. Call initializeDuckDB first.",
@@ -77,7 +83,7 @@ export async function createProjectionTable(
 
     const createTableSQL = `CREATE TABLE IF NOT EXISTS ${tableName} (${columnsSQL})`
 
-    await connection.run(createTableSQL)
+    await connection.exec(createTableSQL)
 
     return {
       success: true,
@@ -97,8 +103,11 @@ export async function createProjectionTable(
  * @param record Record to insert
  * @returns Object indicating success or failure
  */
-export async function insertRecordIntoTable(tableName: string, record: Record<string, unknown>) {
-  if (!instance || !connection) {
+export async function insertRecordIntoTable(
+  tableName: string,
+  record: Record<string, unknown>,
+): Promise<{ success: boolean; message: string }> {
+  if (!db || !connection) {
     return {
       success: false,
       message: "Database not initialized. Call initializeDuckDB first.",
@@ -107,46 +116,35 @@ export async function insertRecordIntoTable(tableName: string, record: Record<st
 
   try {
     const columnNames = Object.keys(record).join(", ")
-    const placeholders = Object.keys(record)
-      .map((_, index) => `$${index + 1}`)
-      .join(", ")
-    const values = Object.values(record)
 
-    const insertSQL = `INSERT INTO ${tableName} (${columnNames}) VALUES (${placeholders})`
-
-    try {
-      const prepared = await connection.prepare(insertSQL)
+    // Format values according to their type
+    const formattedValues = Object.values(record)
+      .map((value) => {
+        if (value === null) {
+          return "NULL"
+        }
 
-      // Bind values based on their types
-      values.forEach((value, index) => {
-        const paramIndex = index + 1
         if (typeof value === "string") {
-          prepared.bindVarchar(paramIndex, value)
-        } else if (typeof value === "number") {
-          if (Number.isInteger(value)) {
-            prepared.bindInteger(paramIndex, value)
-          } else {
-            prepared.bindDouble(paramIndex, value)
-          }
-        } else if (typeof value === "boolean") {
-          prepared.bindBoolean(paramIndex, value)
-        } else if (value === null) {
-          prepared.bindNull(paramIndex)
-        } else if (typeof value === "bigint") {
-          prepared.bindBigInt(paramIndex, value)
-        } else {
-          // For other types, convert to string
-          prepared.bindVarchar(paramIndex, String(value))
+          // Escape single quotes in strings
+          return `'${String(value).replace(/'/g, "''")}'`
         }
+
+        if (typeof value === "boolean") {
+          return value ? "TRUE" : "FALSE"
+        }
+
+        if (value instanceof Date) {
+          return `'${value.toISOString()}'`
+        }
+
+        return String(value)
       })
+      .join(", ")
+
+    const insertSQL = `INSERT INTO ${tableName} (${columnNames}) VALUES (${formattedValues})`
+
+    await connection.exec(insertSQL)
 
-      await prepared.run()
-    } catch (error) {
-      return {
-        success: false,
-        message: `Failed to insert record into ${tableName}: ${error}`,
-      }
-    }
     return {
       success: true,
       message: `Record inserted into ${tableName} successfully`,
@@ -164,8 +162,10 @@ export async function insertRecordIntoTable(tableName: string, record: Record<st
  * @param query SQL query to execute
  * @returns Object with query results or error
  */
-export async function queryDatabase(query: string) {
-  if (!instance || !connection) {
+export async function queryDatabase(
+  query: string,
+): Promise<{ success: boolean; message: string; results: unknown[] | null }> {
+  if (!db || !connection) {
     return {
       success: false,
       message: "Database not initialized. Call initializeDuckDB first.",
@@ -174,11 +174,12 @@ export async function queryDatabase(query: string) {
   }
 
   try {
-    const result = await connection.run(query)
+    const rows = await connection.all(query)
+
     return {
       success: true,
       message: "Query executed successfully",
-      results: result,
+      results: rows,
     }
   } catch (error) {
     return {
@@ -194,7 +195,7 @@ export async function queryDatabase(query: string) {
  * @returns The database instance or null if not initialized
  */
 export function getDatabaseInstance() {
-  return instance
+  return db
 }
 
 /**

src/duckdb/projector.ts

@@ -26,18 +26,33 @@ export async function createEventProjector(name: string, absoluteFilePath: strin
   // Create a unique ID for the projector
   const projectorId = `${name}-${Date.now()}`
 
-  const projectorFunction = await import(absoluteFilePath)
+  try {
+    const projectorFunction = await import(absoluteFilePath)
 
-  // Store the projector info
-  projectors[name] = {
-    name,
-    function: projectorFunction.default as ProjectionFunction,
-  }
+    if (!projectorFunction.default && projectorFunction.projectEvent) {
+      // Store the projector info
+      projectors[name] = {
+        name,
+        function: projectorFunction.projectEvent as ProjectionFunction,
+      }
+    } else if (projectorFunction.default) {
+      // Store the projector info
+      projectors[name] = {
+        name,
+        function: projectorFunction.default as ProjectionFunction,
+      }
+    } else {
+      throw new Error(`No valid projector function found in ${absoluteFilePath}`)
+    }
 
-  return {
-    success: true,
-    projectorId,
-    name,
+    return {
+      success: true,
+      projectorId,
+      name,
+    }
+  } catch (error) {
+    console.error(`Failed to create event projector '${name}' from ${absoluteFilePath}:`, error)
+    throw error
   }
 }
 
@@ -47,7 +62,8 @@ export async function createEventProjector(name: string, absoluteFilePath: strin
  * @returns The projector or undefined if not found
  */
 export function getProjectorByName(name: string): ProjectorInfo | undefined {
-  return projectors[name]
+  const projector = projectors[name]
+  return projector
 }
 
 /**

src/duckdb/stream.ts

@@ -9,9 +9,11 @@ dayjs.extend(utc)
 // Store active streams
 const activeStreams: Record<string, StreamInfo> = {}
 
+const activeDataPumps: Record<string, FlowcoreDataPump> = {}
+
 interface StreamInfo {
   id: string
-  status: "INITIALIZING" | "RUNNING" | "COMPLETED" | "ERROR"
+  status: "INITIALIZING" | "RUNNING" | "COMPLETED" | "ERROR" | "STOPPED"
   eventCount: number
   tenant: string
   dataCore: string
@@ -104,11 +106,10 @@ async function processEventStream(getBearerToken: () => Promise<string>, streamI
 
   try {
     streamInfo.status = "RUNNING"
-
     const startState = getState(streamInfo.startDate)
     const endState = getState(streamInfo.endDate)
 
-    const dataPump = FlowcoreDataPump.create({
+    activeDataPumps[streamId] = FlowcoreDataPump.create({
       auth: {
         getBearerToken: async () => await getBearerToken(),
       },
@@ -133,10 +134,21 @@ async function processEventStream(getBearerToken: () => Promise<string>, streamI
             return
           }
 
+          let successCount = 0
+          let errorCount = 0
+
           for (const event of events) {
-            const projectionResult = await projectEvent(event, streamInfo.targetTable, streamInfo.projectorName)
-            if (projectionResult.success) {
-              streamInfo.eventCount++
+            try {
+              const projectionResult = await projectEvent(event, streamInfo.targetTable, streamInfo.projectorName)
+
+              if (projectionResult.success) {
+                successCount++
+                streamInfo.eventCount++
+              } else {
+                errorCount++
+              }
+            } catch (error) {
+              errorCount++
             }
           }
         },
@@ -148,7 +160,7 @@ async function processEventStream(getBearerToken: () => Promise<string>, streamI
       logger: console,
     })
 
-    await dataPump.start((error?: Error) => {
+    await activeDataPumps[streamId].start((error?: Error) => {
       if (error) {
         streamInfo.status = "ERROR"
         streamInfo.error = `${error}`
@@ -163,6 +175,28 @@ async function processEventStream(getBearerToken: () => Promise<string>, streamI
   }
 }
 
+export function stopEventStreamProjection(streamId: string) {
+  const streamInfo = activeStreams[streamId]
+
+  if (!streamInfo) {
+    throw new Error(`Stream ${streamId} not found`)
+  }
+
+  if (!activeDataPumps[streamId]) {
+    throw new Error(`Data pump for stream ${streamId} not found`)
+  }
+
+  streamInfo.status = "STOPPED"
+
+  activeDataPumps[streamId].stop()
+}
+
+export function stopAllEventStreamProjections() {
+  for (const streamId in activeStreams) {
+    stopEventStreamProjection(streamId)
+  }
+}
+
 function getState(dateString: string) {
   const date = dayjs(dateString)
   return {