Stream processing example: tar-stream -> openpgp -> fs.createWriteStream

tarStreaming.ts

import fs from 'node:fs'
import path from 'node:path'
import stream from 'node:stream'
import { ReadableStream } from 'node:stream/web'

import * as tar from 'tar-stream'
import * as openpgp from 'openpgp'

const files = [
  '/path/to/file1',
  '/path/to/file2',
  // ...
]

const dest = '/path/to/output.tar.pgp'

const publicKey = '/path/to/pgp_public_key.asc'

// tar pack streaming
// ===============

const pack = tar.pack()

// web stream for openpgp v6
const tarReadableStream = new ReadableStream<Uint8Array>({
  start(controller) {
    pack.on('data', (chunk: Buffer) => {
      console.log('chunk', chunk.byteLength)

      // Buffer to Uint8Array
      const uint8Array = new Uint8Array(chunk.buffer, chunk.byteOffset, chunk.byteLength)
      controller.enqueue(uint8Array)
      if (controller.desiredSize || 0 <= 0) pack.pause()
    })

    pack.on('error', (err) => controller.error(err))
    pack.on('end', () => controller.close())
  },
  pull() {
    pack.resume()
  },
  cancel() {
    pack.destroy()
  },
})

// or node stream for openpgp v5
class TarReadable extends stream.Readable {
  pack: tar.Pack

  constructor(pack: tar.Pack) {
    super()
    this.pack = pack

    pack.on('data', (chunk: Buffer) => {
      console.log('chunk', chunk.byteLength)
      if (!this.push(chunk)) this.pause()
    })
    pack.on('end', () => this.push(null))
    pack.on('error', (err) => this.destroy(err))
  }

  _read() {
    this.pack.resume()
  }

  _destroy(error: Error | null, callback: (error?: Error | null) => void): void {
    this.pack.destroy(error || undefined)
    callback(error)
  }
}

const tarReadable = new TarReadable(pack)

// entries 要完成一個之後才添加下一個
// 不能包成 await Promise 等 entries 添加完, 它需要 pipe 到 write stream 寫完後才算完成一個
// 在 await 之後才 pipe write stream 會變成 promise 死鎖
function addEntries(files: string[], index = 0) {
  const file = files[index]
  if (!file) return pack.finalize()

  console.log(index, 'add', file)
  const { size } = fs.statSync(file)
  const name = path.basename(file)

  const entry = pack.entry({ name, size }, (err) => {
    console.log(index, 'done', file)
    if (err) throw err
    addEntries(files, index + 1)
  })

  fs.createReadStream(file).pipe(entry)
}

console.log('add entries')
addEntries(files)

// PGP encrypt streaming
// ===============

const pubkey = fs.readFileSync(publicKey, 'utf8')
const pgpkey = await openpgp.readKey({ armoredKey: pubkey })
const message = await openpgp.createMessage({ binary: tarReadableStream })

// or node:stream Readable
// const message = await openpgp.createMessage({ binary: tarReadable })

const encrypt = await openpgp.encrypt({
  message,
  encryptionKeys: pgpkey,
  format: 'binary',
})

// pipe write stream
// ===============

const writeStream = fs.createWriteStream(dest)

console.log('pipe writable')
await stream.promises.pipeline(
  encrypt,
  writeStream,
)

console.log('completed')