use to

scripts/.env.sample

@@ -21,3 +21,13 @@ export START_HOURS_AGO="72"
 export END_HOURS_AGO="0"
 export HOUR_WINDOW="1"
 export TAGS_TO_CONVERT="address"
+
+export NEW_MEASUREMENT_VERSION=v1
+
+#export MEASUREMENT="failed_txn_event"
+#export TAGS_TO_KEEP="network"
+#export FIELDS="signature,errorDetails"
+export MEASUREMENT="txn_event"
+export TAGS_TO_KEEP="ecosystem,network,eventCategory"
+export FIELDS="signature,amount,eventDetails,address,claimant"
+

scripts/influxdb.fixer.ts

@@ -16,13 +16,15 @@ const INFLUX_BUCKET = envOrErr('INFLUX_BUCKET')
 
 const MEASUREMENT = process.env.MEASUREMENT ?? 'txn_event'
 const NEW_MEASUREMENT_VERSION = process.env.NEW_MEASUREMENT_VERSION ?? 'v1'
+const TAGS_TO_KEEP =
+  process.env.TAGS_TO_KEEP ?? 'ecosystem,network,eventCategory'
+const FIELDS =
+  process.env.FIELDS ?? 'signature,amount,eventDetails,address,claimant'
 
 const START_HOURS_AGO = process.env.START_HOURS_AGO ?? '72' // 3 days
 const END_HOURS_AGO = process.env.END_HOURS_AGO ?? '0' // now
 const HOUR_WINDOW = process.env.HOUR_WINDOW ?? '1'
 
-const TAGS_TO_CONVERT = process.env.TAGS_TO_CONVERT ?? 'address'
-
 async function main() {
   console.log(
     `Connecting to influx <${INFLUX_URL}/${INFLUX_ORG}/${INFLUX_BUCKET}>`
@@ -34,7 +36,8 @@ async function main() {
 
   const influxDB = new InfluxDB({ url: INFLUX_URL, token: INFLUX_TOKEN })
   const readApi = influxDB.getQueryApi(INFLUX_ORG)
-  const tagsToConvert = TAGS_TO_CONVERT.split(',')
+  const tagsToKeep = TAGS_TO_KEEP.split(',')
+  const fields = FIELDS.split(',')
 
   let startHoursAgo = Number.parseInt(START_HOURS_AGO)
   const endHoursAgo = Number.parseInt(END_HOURS_AGO)
@@ -49,7 +52,8 @@ async function main() {
         INFLUX_BUCKET,
         MEASUREMENT,
         NEW_MEASUREMENT_VERSION,
-        tagsToConvert,
+        tagsToKeep,
+        fields,
         currentStartHoursAgo,
         currentEndHoursago,
         readApi
@@ -69,22 +73,28 @@ async function mapAndWrite(
   bucket: string,
   measurement: string,
   newMeasurementVersion: string,
-  tagsToConvert: string[],
+  tagsToKeep: string[],
+  fields: string[],
   start: string,
   end: string,
   readApi: QueryApi
 ): Promise<void> {
   const stream = `from(bucket: "${bucket}")
   |> range(start: ${start}, stop: ${end})
-  |> filter(fn: (r) => r._measurement == "${measurement}")`
+  |> filter(fn: (r) => r._measurement == "${measurement}")
+  |> pivot(rowKey:["_time"], columnKey: ["_field"], valueColumn: "_value")`
 
   const mapper = `${stream}
-  |> map(fn: (r) => ({ r with _measurement: "${measurement}_${newMeasurementVersion}", ${tagsToConvert
-    .map((tag) => `${tag}: r.${tag}`)
-    .join(', ')} }))
-  |> drop(columns: [${tagsToConvert.map((tag) => `"${tag}"`).join(',')}])
-  |> to(bucket: "${bucket}")
-  `
+  |> set(key: "_measurement", value: "${measurement}_${newMeasurementVersion}")
+  |> to(
+      bucket: "${bucket}",
+      timeColumn: "_time",
+      tagColumns: [${tagsToKeep.map((tag) => `"${tag}"`).join(',')}],
+      fieldFn: (r) => ({ ${fields
+        .map((field) => `${field}: r.${field}`)
+        .join(',')} })
+    )`
+
   console.log(` Executing the following query: ${mapper}`)
   const response = await readApi.queryRaw(mapper)
   console.log(response)