fix: orphan object client detector

fenos · fenos · commit e58e9a693c84 · 2025-04-21T16:36:07.000+02:00
diff --git a/src/internal/streams/ndjson.ts b/src/internal/streams/ndjson.ts
@@ -0,0 +1,56 @@
+import { Transform, TransformCallback } from 'stream'
+import { StringDecoder } from 'string_decoder'
+
+export class NdJsonTransform extends Transform {
+  private decoder = new StringDecoder('utf8')
+  private buffer = ''
+
+  constructor() {
+    super({ readableObjectMode: true })
+  }
+
+  _transform(chunk: Buffer, encoding: BufferEncoding, callback: TransformCallback) {
+    // decode safely across chunk boundaries
+    this.buffer += this.decoder.write(chunk)
+
+    let newlineIdx: number
+    while ((newlineIdx = this.buffer.indexOf('\n')) !== -1) {
+      const line = this.buffer.slice(0, newlineIdx)
+      this.buffer = this.buffer.slice(newlineIdx + 1)
+      if (line.trim()) {
+        let obj
+        try {
+          obj = JSON.parse(line)
+        } catch (err) {
+          if (err instanceof Error) {
+            // this is the case when JSON.parse fails
+            return callback(new Error(`Invalid JSON on flush: ${err.message}`))
+          }
+
+          return callback(err as Error)
+        }
+        // .push() participates in backpressure automatically
+        this.push(obj)
+      }
+    }
+
+    callback()
+  }
+
+  _flush(callback: TransformCallback) {
+    this.buffer += this.decoder.end()
+    if (this.buffer.trim()) {
+      try {
+        this.push(JSON.parse(this.buffer))
+      } catch (err) {
+        if (err instanceof Error) {
+          // this is the case when JSON.parse fails
+          return callback(new Error(`Invalid JSON on flush: ${err.message}`))
+        }
+
+        return callback(err as Error)
+      }
+    }
+    callback()
+  }
+}
diff --git a/src/scripts/orphan-client.ts b/src/scripts/orphan-client.ts
@@ -0,0 +1,175 @@
+import axios from 'axios'
+import { NdJsonTransform } from '@internal/streams/ndjson'
+import fs from 'fs'
+import path from 'path'
+
+const ADMIN_URL = process.env.ADMIN_URL
+const ADMIN_API_KEY = process.env.ADMIN_API_KEY
+const TENANT_ID = process.env.TENANT_ID
+const BUCKET_ID = process.env.BUCKET_ID
+
+const BEFORE = undefined // new Date().toISOString()
+
+const FILE_PATH = (operation: string) =>
+  `../../dist/${operation}-${TENANT_ID}-${Date.now()}-orphan-objects.json`
+
+const client = axios.create({
+  baseURL: ADMIN_URL,
+  headers: {
+    ApiKey: ADMIN_API_KEY,
+  },
+})
+
+interface OrphanObject {
+  event: 'data'
+  type: 's3Orphans'
+  value: {
+    name: string
+    version: string
+    size: number
+  }[]
+}
+
+interface PingObject {
+  event: 'ping'
+}
+
+async function main() {
+  const action = process.argv[2]
+
+  if (!action) {
+    console.error('Please provide an action: list or delete')
+    return
+  }
+
+  if (action === 'list') {
+    await listOrphans(TENANT_ID, BUCKET_ID)
+    return
+  }
+
+  await deleteS3Orphans(TENANT_ID, BUCKET_ID)
+}
+
+/**
+ * List Orphan objects in a bucket
+ * @param tenantId
+ * @param bucketId
+ */
+async function listOrphans(tenantId: string, bucketId: string) {
+  const request = await client.get(`/tenants/${tenantId}/buckets/${bucketId}/orphan-objects`, {
+    responseType: 'stream',
+    params: {
+      before: BEFORE,
+    },
+  })
+
+  const transformStream = new NdJsonTransform()
+  request.data.on('error', (err: Error) => {
+    transformStream.emit('error', err)
+  })
+
+  const jsonStream = request.data.pipe(transformStream)
+
+  await writeStreamToJsonArray(jsonStream, FILE_PATH('list'))
+}
+
+/**
+ * Deletes S3 orphan objects in a bucket
+ * @param tenantId
+ * @param bucketId
+ */
+async function deleteS3Orphans(tenantId: string, bucketId: string) {
+  const request = await client.delete(`/tenants/${tenantId}/buckets/${bucketId}/orphan-objects`, {
+    responseType: 'stream',
+    data: {
+      deleteS3Keys: true,
+      before: BEFORE,
+    },
+  })
+
+  const transformStream = new NdJsonTransform()
+  request.data.on('error', (err: Error) => {
+    transformStream.emit('error', err)
+  })
+
+  const jsonStream = request.data.pipe(transformStream)
+
+  await writeStreamToJsonArray(jsonStream, FILE_PATH('delete'))
+}
+
+/**
+ * Writes the output to a JSON array
+ * @param stream
+ * @param relativePath
+ */
+async function writeStreamToJsonArray(
+  stream: NodeJS.ReadableStream,
+  relativePath: string
+): Promise<void> {
+  const filePath = path.resolve(__dirname, relativePath)
+  const localFile = fs.createWriteStream(filePath)
+
+  // Start with an empty array
+  localFile.write('[\n')
+  let isFirstItem = true
+
+  return new Promise((resolve, reject) => {
+    let receivedAnyData = false
+
+    stream.on('data', (data: OrphanObject | PingObject) => {
+      if (data.event === 'ping') {
+        console.log('Received ping event, ignoring')
+        return
+      }
+
+      if (data.event === 'data' && data.value && Array.isArray(data.value)) {
+        receivedAnyData = true
+        console.log(`Processing ${data.value.length} objects`)
+
+        for (const item of data.value) {
+          if (!isFirstItem) {
+            localFile.write(',\n')
+          } else {
+            isFirstItem = false
+          }
+
+          localFile.write(JSON.stringify(item, null, 2))
+        }
+      } else {
+        console.warn(
+          'Received data with invalid format:',
+          JSON.stringify(data).substring(0, 100) + '...'
+        )
+      }
+    })
+
+    stream.on('error', (err) => {
+      console.error('Stream error:', err)
+      localFile.end('\n]', () => {
+        reject(err)
+      })
+    })
+
+    stream.on('end', () => {
+      localFile.write('\n]')
+      localFile.end(() => {
+        resolve()
+      })
+
+      if (!receivedAnyData) {
+        console.warn(`No data was received! File might be empty: ${filePath}`)
+      } else {
+        // Check if the file exists and has content
+        console.log(`Finished writing data to ${filePath}. Data was received and saved.`)
+      }
+    })
+  })
+}
+
+main()
+  .catch((e) => {
+    console.error('Error:', e)
+  })
+  .then(() => {
+    console.log('Done')
+  })
diff --git a/src/test/ndjson.test.ts b/src/test/ndjson.test.ts
@@ -0,0 +1,84 @@
+// NdJsonTransform.test.ts
+
+import { Buffer } from 'buffer'
+import { NdJsonTransform } from '@internal/streams/ndjson'
+
+/**
+ * Helper that writes the given chunks into the transform,
+ * collects all the parsed objects (in order), and resolves
+ * them—or rejects on error.
+ */
+function collect(transform: NdJsonTransform, chunks: Buffer[]): Promise<any[]> {
+  return new Promise((resolve, reject) => {
+    const out: any[] = []
+    transform.on('data', (obj) => out.push(obj))
+    transform.on('error', (err) => reject(err))
+    transform.on('end', () => resolve(out))
+    for (const c of chunks) transform.write(c)
+    transform.end()
+  })
+}
+
+describe('NdJsonTransform', () => {
+  it('parses a single JSON object terminated by newline', async () => {
+    const t = new NdJsonTransform()
+    const result = await collect(t, [Buffer.from('{"foo":123}\n')])
+    expect(result).toEqual([{ foo: 123 }])
+  })
+
+  it('parses multiple JSON objects in one chunk', async () => {
+    const t = new NdJsonTransform()
+    const chunk = Buffer.from('{"a":1}\n{"b":2}\n')
+    const result = await collect(t, [chunk])
+    expect(result).toEqual([{ a: 1 }, { b: 2 }])
+  })
+
+  it('skips empty and whitespace-only lines', async () => {
+    const t = new NdJsonTransform()
+    const chunk = Buffer.from('\n   \n{"x":10}\n  \n')
+    const result = await collect(t, [chunk])
+    expect(result).toEqual([{ x: 10 }])
+  })
+
+  it('parses JSON split across multiple chunks', async () => {
+    const t = new NdJsonTransform()
+    const chunks = [Buffer.from('{"split":'), Buffer.from('true}\n')]
+    const result = await collect(t, chunks)
+    expect(result).toEqual([{ split: true }])
+  })
+
+  it('parses final line without trailing newline on flush', async () => {
+    const t = new NdJsonTransform()
+    const chunks = [Buffer.from('{"end":"last"}')]
+    const result = await collect(t, chunks)
+    expect(result).toEqual([{ end: 'last' }])
+  })
+
+  it('propagates parse errors in _transform (invalid JSON with newline)', async () => {
+    const t = new NdJsonTransform()
+    const bad = Buffer.from('{"foo": bad}\n')
+    await expect(collect(t, [bad])).rejects.toThrow(/Invalid JSON on flush:/)
+  })
+
+  it('propagates parse errors in _flush (invalid final JSON)', async () => {
+    const t = new NdJsonTransform()
+    const bad = Buffer.from('{"incomplete":123')
+    await expect(collect(t, [bad])).rejects.toThrow(/Invalid JSON on flush:/)
+  })
+
+  it('handles multi-byte UTF-8 characters split across chunk boundary', async () => {
+    const t = new NdJsonTransform()
+    const full = Buffer.from('{"emoji":"💩"}\n', 'utf8')
+    // Split in the middle of the 4‑byte 💩 codepoint:
+    const chunk1 = full.slice(0, 12) // up through two bytes of the emoji
+    const chunk2 = full.slice(12) // remainder of emoji + '}' + '\n'
+    const result = await collect(t, [chunk1, chunk2])
+    expect(result).toEqual([{ emoji: '💩' }])
+  })
+
+  it('emits no data for completely empty input', async () => {
+    const t = new NdJsonTransform()
+    const result = await collect(t, [])
+    expect(result).toEqual([])
+  })
+})
