support historical event export

Author: yakkomajuri

index.ts

@@ -5,7 +5,6 @@ import { Client } from 'pg'
 type PostgresPlugin = Plugin<{
     global: {
         pgClient: Client
-        buffer: ReturnType<typeof createBuffer>
         eventsToIgnore: Set<string>
         sanitizedTableName: string
     }
@@ -17,37 +16,54 @@ type PostgresPlugin = Plugin<{
         tableName: string
         dbUsername: string
         dbPassword: string
-        uploadSeconds: string
-        uploadMegabytes: string
         eventsToIgnore: string
-        isHeroku: 'Yes' | 'No'
+        hasSelfSignedCert: 'Yes' | 'No'
     }
 }>
 
 type PostgresMeta = PluginMeta<PostgresPlugin>
 
 interface ParsedEvent {
-    uuid: string
+    uuid?: string
     eventName: string
-    properties: Record<string, any>
-    elements: Record<string, any>
-    set: Record<string, any>
-    set_once: Record<string, any>
+    properties: string
+    elements: string
+    set: string
+    set_once: string
     distinct_id: string
     team_id: number
-    ip: string
+    ip: string | null
     site_url: string
     timestamp: string
 }
 
-type InsertQueryValue = string | number
-
 interface UploadJobPayload {
     batch: ParsedEvent[]
     batchId: number
     retriesPerformedSoFar: number
 }
 
+const randomBytes = (): string => {
+    return (((1 + Math.random()) * 0x10000) | 0).toString(16).substring(1)
+}
+
+const generateUuid = (): string => {
+    return (
+        randomBytes() +
+        randomBytes() +
+        '-' +
+        randomBytes() +
+        '-3' +
+        randomBytes().substr(0, 2) +
+        '-' +
+        randomBytes() +
+        '-' +
+        randomBytes() +
+        randomBytes() +
+        randomBytes()
+    ).toLowerCase()
+}
+
 export const jobs: PostgresPlugin['jobs'] = {
     uploadBatchToPostgres: async (payload: UploadJobPayload, meta: PostgresMeta) => {
         await insertBatchIntoPostgres(payload, meta)
@@ -66,9 +82,6 @@ export const setupPlugin: PostgresPlugin['setupPlugin'] = async (meta) => {
         }
     }
 
-    const uploadMegabytes = Math.max(1, Math.min(parseInt(config.uploadMegabytes) || 1, 10))
-    const uploadSeconds = Math.max(1, Math.min(parseInt(config.uploadSeconds) || 1, 600))
-
     global.sanitizedTableName = sanitizeSqlIdentifier(config.tableName)
 
     const queryError = await executeQuery(
@@ -93,103 +106,102 @@ export const setupPlugin: PostgresPlugin['setupPlugin'] = async (meta) => {
         throw new Error(`Unable to connect to PostgreSQL instance and create table with error: ${queryError.message}`)
     }
 
-    global.buffer = createBuffer({
-        limit: uploadMegabytes * 1024 * 1024,
-        timeoutSeconds: uploadSeconds,
-        onFlush: async (batch) => {
-            await insertBatchIntoPostgres(
-                { batch, batchId: Math.floor(Math.random() * 1000000), retriesPerformedSoFar: 0 },
-                meta
-            )
-        },
-    })
-
     global.eventsToIgnore = new Set(
         config.eventsToIgnore ? config.eventsToIgnore.split(',').map((event) => event.trim()) : null
     )
 }
 
-export async function onEvent(event: PluginEvent, { global }: PostgresMeta) {
-    const {
-        event: eventName,
-        properties,
-        $set,
-        $set_once,
-        distinct_id,
-        team_id,
-        site_url,
-        now,
-        sent_at,
-        uuid,
-        ..._discard
-    } = event
-
-    const ip = properties?.['$ip'] || event.ip
-    const timestamp = event.timestamp || properties?.timestamp || now || sent_at
-    let ingestedProperties = properties
-    let elements = []
-
-    // only move prop to elements for the $autocapture action
-    if (eventName === '$autocapture' && properties && '$elements' in properties) {
-        const { $elements, ...props } = properties
-        ingestedProperties = props
-        elements = $elements
-    }
+export async function exportEvents(events: PluginEvent[], { global, jobs }: PostgresMeta) {
+    const batch: ParsedEvent[] = []
+    for (const event of events) {
+        const {
+            event: eventName,
+            properties,
+            $set,
+            $set_once,
+            distinct_id,
+            team_id,
+            site_url,
+            now,
+            sent_at,
+            uuid,
+            ..._discard
+        } = event
+
+        if (global.eventsToIgnore.has(eventName)) {
+            continue
+        }
+
+        const ip = properties?.['$ip'] || event.ip
+        const timestamp = event.timestamp || properties?.timestamp || now || sent_at
+        let ingestedProperties = properties
+        let elements = []
+
+        // only move prop to elements for the $autocapture action
+        if (eventName === '$autocapture' && properties && '$elements' in properties) {
+            const { $elements, ...props } = properties
+            ingestedProperties = props
+            elements = $elements
+        }
 
-    const parsedEvent = {
-        uuid,
-        eventName,
-        properties: JSON.stringify(ingestedProperties || {}),
-        elements: JSON.stringify(elements || {}),
-        set: JSON.stringify($set || {}),
-        set_once: JSON.stringify($set_once || {}),
-        distinct_id,
-        team_id,
-        ip,
-        site_url,
-        timestamp: new Date(timestamp).toISOString(),
+        const parsedEvent: ParsedEvent = {
+            uuid,
+            eventName,
+            properties: JSON.stringify(ingestedProperties || {}),
+            elements: JSON.stringify(elements || {}),
+            set: JSON.stringify($set || {}),
+            set_once: JSON.stringify($set_once || {}),
+            distinct_id,
+            team_id,
+            ip,
+            site_url,
+            timestamp: new Date(timestamp).toISOString(),
+        }
+
+        batch.push(parsedEvent)
     }
 
-    if (!global.eventsToIgnore.has(eventName)) {
-        global.buffer.add(parsedEvent)
+    if (batch.length > 0) {
+        await jobs
+            .uploadBatchToPostgres({ batch, batchId: Math.floor(Math.random() * 1000000), retriesPerformedSoFar: 0 })
+            .runNow()
     }
 }
 
 export const insertBatchIntoPostgres = async (payload: UploadJobPayload, { global, jobs, config }: PostgresMeta) => {
-    let values: InsertQueryValue[] = []
+    let values: any[] = []
     let valuesString = ''
 
     for (let i = 0; i < payload.batch.length; ++i) {
         const { uuid, eventName, properties, elements, set, set_once, distinct_id, team_id, ip, site_url, timestamp } =
             payload.batch[i]
 
+
         // Creates format: ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11), ($12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22)
         valuesString += ' ('
         for (let j = 1; j <= 11; ++j) {
             valuesString += `$${11 * i + j}${j === 11 ? '' : ', '}`
         }
         valuesString += `)${i === payload.batch.length - 1 ? '' : ','}`
-
-        values = [
-            ...values,
-            ...[
-                uuid,
-                eventName,
-                properties,
-                elements,
-                set,
-                set_once,
-                distinct_id,
-                team_id,
-                ip,
-                site_url,
-                timestamp,
-            ],
-        ]
+        
+        values = values.concat([
+            uuid || generateUuid(),
+            eventName,
+            properties,
+            elements,
+            set,
+            set_once,
+            distinct_id,
+            team_id,
+            ip,
+            site_url,
+            timestamp,
+        ])
     }
 
     console.log(
-        `(Batch Id: ${payload.batchId}) Flushing ${payload.batch.length} event${payload.batch.length > 1 ? 's' : ''
+        `(Batch Id: ${payload.batchId}) Flushing ${payload.batch.length} event${
+            payload.batch.length > 1 ? 's' : ''
         } to Postgres instance`
     )
 
@@ -217,42 +229,38 @@ export const insertBatchIntoPostgres = async (payload: UploadJobPayload, { globa
 }
 
 const executeQuery = async (query: string, values: any[], config: PostgresMeta['config']): Promise<Error | null> => {
-    const basicConnectionOptions = config.databaseUrl ? {
-        connectionString: config.databaseUrl
-    } : {
-        user: config.dbUsername,
-        password: config.dbPassword,
-        host: config.host,
-        database: config.dbName,
-        port: parseInt(config.port),
-    }
-    const pgClient = new Client(
-        {
-            ...basicConnectionOptions,
-            ssl: {
-                rejectUnauthorized: config.isHeroku === "No"
-            }
-        }
-    )
+    const basicConnectionOptions = config.databaseUrl
+        ? {
+              connectionString: config.databaseUrl,
+          }
+        : {
+              user: config.dbUsername,
+              password: config.dbPassword,
+              host: config.host,
+              database: config.dbName,
+              port: parseInt(config.port),
+          }
+    const pgClient = new Client({
+        ...basicConnectionOptions,
+        ssl: {
+            rejectUnauthorized: config.hasSelfSignedCert === 'No',
+        },
+    })
 
     await pgClient.connect()
 
     let error: Error | null = null
     try {
         await pgClient.query(query, values)
     } catch (err) {
-        error = err
+        error = err as Error
     }
 
     await pgClient.end()
 
     return error
 }
 
-export const teardownPlugin: PostgresPlugin['teardownPlugin'] = ({ global }) => {
-    global.buffer.flush()
-}
-
 const sanitizeSqlIdentifier = (unquotedIdentifier: string): string => {
     return unquotedIdentifier.replace(/[^\w\d_]+/g, '')
 }

package.json

@@ -1,7 +1,7 @@
 {
     "name": "@posthog/postgres-plugin",
     "private": true,
-    "version": "0.0.1",
+    "version": "0.1.0",
     "description": "Export PostHog events to a PostgreSQL instance on ingestion.",
     "devDependencies": {
         "@posthog/plugin-contrib": "^0.0.3",