From 57bcd7d5d5faf4314a178cf2a1463d8bd23d2c4a Mon Sep 17 00:00:00 2001
From: Daniele Bacarella <daniele.bacarella@gmail.com>
Date: Sat, 11 May 2024 14:37:56 +0200
Subject: [PATCH] - Added optional 'level' property to TransportPipelineOptions
 interface - A level can now be defined for pipelines defined inside 'targets'
 - Added UT in 'pipeline.test.js' to check expected behaviour with 'dedupe'

---
 lib/transport.js                |  1 +
 lib/worker.js                   |  8 +++-
 pino.d.ts                       |  1 +
 test/transport/pipeline.test.js | 68 +++++++++++++++++++++++++++------
 4 files changed, 66 insertions(+), 12 deletions(-)

diff --git a/lib/transport.js b/lib/transport.js
index ee480abf9..b8ce347f2 100644
--- a/lib/transport.js
+++ b/lib/transport.js
@@ -97,6 +97,7 @@ function transport (fullOptions) {
       return dest.pipeline.map((t) => {
         return {
           ...t,
+          level: dest.level, // duplicate the pipeline `level` property defined in the upper level
           target: fixTarget(t.target)
         }
       })
diff --git a/lib/worker.js b/lib/worker.js
index a0864b765..dcbd5d391 100644
--- a/lib/worker.js
+++ b/lib/worker.js
@@ -94,15 +94,21 @@ module.exports = async function ({ targets, pipelines, levels, dedupe }) {
   if (pipelines && pipelines.length) {
     pipelines = await Promise.all(
       pipelines.map(async (p) => {
+        let level
         const pipeDests = await Promise.all(
           p.map(async (t) => {
+            // level assigned to pipeline is duplicated over all its targets, just store it
+            level = t.level
             const fn = await loadTransportStreamBuilder(t.target)
             const stream = await fn(t.options)
             return stream
           }
           ))
 
-        return { stream: createPipeline(pipeDests) }
+        return {
+          level,
+          stream: createPipeline(pipeDests)
+        }
       })
     )
     targetStreams.push(...pipelines)
diff --git a/pino.d.ts b/pino.d.ts
index 33f9788ca..3efe72276 100644
--- a/pino.d.ts
+++ b/pino.d.ts
@@ -259,6 +259,7 @@ declare namespace pino {
 
     interface TransportPipelineOptions<TransportOptions = Record<string, any>> extends TransportBaseOptions<TransportOptions>{
         pipeline: TransportSingleOptions<TransportOptions>[]
+        level?: LevelWithSilentOrString
     }
 
     interface TransportMultiOptions<TransportOptions = Record<string, any>> extends TransportBaseOptions<TransportOptions>{
diff --git a/test/transport/pipeline.test.js b/test/transport/pipeline.test.js
index 14d435ae5..a845d4425 100644
--- a/test/transport/pipeline.test.js
+++ b/test/transport/pipeline.test.js
@@ -6,6 +6,7 @@ const { readFile } = require('fs').promises
 const { watchFileCreated, file } = require('../helper')
 const { test } = require('tap')
 const pino = require('../../')
+const { DEFAULT_LEVELS } = require('../../lib/constants')
 
 const { pid } = process
 const hostname = os.hostname()
@@ -29,21 +30,15 @@ test('pino.transport with a pipeline', async ({ same, teardown }) => {
   same(result, {
     pid,
     hostname,
-    level: 30,
+    level: DEFAULT_LEVELS.info,
     msg: 'hello',
     service: 'pino' // this property was added by the transform
   })
 })
 
 test('pino.transport with targets containing pipelines', async ({ same, teardown }) => {
-  const destinationA = join(
-    os.tmpdir(),
-    '_' + Math.random().toString(36).substr(2, 9)
-  )
-  const destinationB = join(
-    os.tmpdir(),
-    '_' + Math.random().toString(36).substr(2, 9)
-  )
+  const destinationA = file()
+  const destinationB = file()
   const transport = pino.transport({
     targets: [
       {
@@ -76,14 +71,65 @@ test('pino.transport with targets containing pipelines', async ({ same, teardown
   same(resultA, {
     pid,
     hostname,
-    level: 30,
+    level: DEFAULT_LEVELS.info,
     msg: 'hello'
   })
   same(resultB, {
     pid,
     hostname,
-    level: 30,
+    level: DEFAULT_LEVELS.info,
     msg: 'hello',
     service: 'pino' // this property was added by the transform
   })
 })
+
+test('pino.transport with targets containing pipelines with levels defined and dedupe', async ({ same, teardown }) => {
+  const destinationA = file()
+  const destinationB = file()
+  const transport = pino.transport({
+    targets: [
+      {
+        target: join(__dirname, '..', 'fixtures', 'to-file-transport.js'),
+        options: { destination: destinationA },
+        level: DEFAULT_LEVELS.info
+      },
+      {
+        pipeline: [
+          {
+            target: join(__dirname, '..', 'fixtures', 'transport-transform.js')
+          },
+          {
+            target: join(__dirname, '..', 'fixtures', 'to-file-transport.js'),
+            options: { destination: destinationB }
+          }
+        ],
+        level: DEFAULT_LEVELS.error
+      }
+    ],
+    dedupe: true
+  })
+
+  teardown(transport.end.bind(transport))
+  const instance = pino(transport)
+  instance.info('hello info')
+  instance.error('hello error')
+  await watchFileCreated(destinationA)
+  await watchFileCreated(destinationB)
+  const resultA = JSON.parse(await readFile(destinationA))
+  const resultB = JSON.parse(await readFile(destinationB))
+  delete resultA.time
+  delete resultB.time
+  same(resultA, {
+    pid,
+    hostname,
+    level: DEFAULT_LEVELS.info,
+    msg: 'hello info'
+  })
+  same(resultB, {
+    pid,
+    hostname,
+    level: DEFAULT_LEVELS.error,
+    msg: 'hello error',
+    service: 'pino' // this property was added by the transform
+  })
+})