24474844911

Committed 15 Apr 2026 07:46PM UTC coverage: 79.064% (-3.4%) from 82.473%

Build # 24474844911

Build Type

Pull #1024

github

Committed by

web-flow

Commit Message

Merge 084bba3bd into 2e9e38312

Pull Request Pull Request #1024: fix: use correct JSON Schema keyword in getSignedURLs response

Coverage Stats

3035 of 4026 branches covered (75.38%)

Branch coverage included in aggregate %.

1 of 1 new or added line in 1 file covered. (100.0%)

4301 existing lines in 99 files now uncovered.

30111 of 37897 relevant lines covered (79.45%)

311.78 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

75.73

/src/internal/pubsub/postgres.ts

import EventEmitter from 'node:events'
import { ERRORS } from '@internal/errors'
import { logger, logSchema } from '@internal/monitoring'
import createSubscriber, { Subscriber } from 'pg-listen'
import { PubSubAdapter } from './adapter'

export class PostgresPubSub extends EventEmitter implements PubSubAdapter {
  isConnected = false
  subscriber: Subscriber

  constructor(connectionString: string) {
    super()
    this.subscriber = createSubscriber(
      { connectionString },
      {
        retryInterval: (attempt) => Math.min(attempt * 100, 1000),
        retryTimeout: 60 * 1000 * 60 * 14, // 24h
      }
    )

    this.subscriber.events.on('error', (e) => {
      this.isConnected = false
      this.emit('error', e)
    })
  }

  async start(opts?: { signal?: AbortSignal }): Promise<void> {
    if (opts?.signal?.aborted) {
      throw ERRORS.Aborted('Postgres pubsub connection aborted')
    }

    await this.subscriber.connect()
    this.isConnected = true

    if (opts?.signal) {
      opts.signal.addEventListener(
        'abort',
        async () => {
          logSchema.info(logger, '[PubSub] Stopping', {
            type: 'pubsub',
          })
          await this.close()
        },
        { once: true }
      )
    }

    await Promise.all(
      this.subscriber.notifications.eventNames().map(async (channel) => {
        return this.subscriber.listenTo(channel as string)
      })
    )
  }

  async close(): Promise<void> {
    this.isConnected = false
    this.subscriber.notifications.eventNames().forEach((event) => {
      this.subscriber.notifications.removeAllListeners(event)
    })
    await this.subscriber.close()
    logSchema.info(logger, '[PubSub] Exited', {
      type: 'pubsub',
    })
  }

  async publish(channel: string, payload: unknown): Promise<void> {
    await this.subscriber.notify(channel, payload)
  }

  async subscribe(channel: string, cb: (payload: any) => void): Promise<void> {
    const listenerCount = this.subscriber.notifications.listenerCount(channel)
    this.subscriber.notifications.on(channel, cb)

    if (this.isConnected && listenerCount === 0) {
      await this.subscriber.listenTo(channel)
    }
  }

  async unsubscribe(channel: string, cb: (payload: any) => void): Promise<void> {
    this.subscriber.notifications.removeListener(channel, cb)

    const isListening = this.subscriber.notifications.listenerCount(channel) > 0

    if (!isListening && this.isConnected) {
      await this.subscriber.unlisten(channel)
    }
  }
}

1	import EventEmitter from 'node:events'	1✔
2	import { ERRORS } from '@internal/errors'	1✔
3	import { logger, logSchema } from '@internal/monitoring'	1✔
4	import createSubscriber, { Subscriber } from 'pg-listen'	1✔
5	import { PubSubAdapter } from './adapter'	1✔
6		1✔
7	export class PostgresPubSub extends EventEmitter implements PubSubAdapter {	1✔
8	isConnected = false	1✔
9	subscriber: Subscriber	1✔
10		1✔
11	constructor(connectionString: string) {	1✔
12	super()	40✔
13	this.subscriber = createSubscriber(	40✔
14	{ connectionString },	40✔
15	{	40✔
16	retryInterval: (attempt) => Math.min(attempt * 100, 1000),	40✔
17	retryTimeout: 60 * 1000 * 60 * 14, // 24h	40✔
18	}	40✔
19	)	40✔
20		40✔
21	this.subscriber.events.on('error', (e) => {	40✔
22	this.isConnected = false	×
23	this.emit('error', e)	×
24	})	40✔
25	}	40✔
26		1✔
27	async start(opts?: { signal?: AbortSignal }): Promise<void> {	1✔
28	if (opts?.signal?.aborted) {	2!
29	throw ERRORS.Aborted('Postgres pubsub connection aborted')	×
UNCOV 30	}	×
31		2✔
32	await this.subscriber.connect()	2✔
33	this.isConnected = true	2✔
34		2✔
35	if (opts?.signal) {	2!
36	opts.signal.addEventListener(	×
UNCOV 37	'abort',	×
UNCOV 38	async () => {	×
39	logSchema.info(logger, '[PubSub] Stopping', {	×
UNCOV 40	type: 'pubsub',	×
UNCOV 41	})	×
42	await this.close()	×
UNCOV 43	},	×
UNCOV 44	{ once: true }	×
UNCOV 45	)	×
UNCOV 46	}	×
47		2✔
48	await Promise.all(	2✔
49	this.subscriber.notifications.eventNames().map(async (channel) => {	2✔
50	return this.subscriber.listenTo(channel as string)	×
51	})	2✔
52	)	2✔
53	}	2✔
54		1✔
55	async close(): Promise<void> {	1✔
56	this.isConnected = false	2✔
57	this.subscriber.notifications.eventNames().forEach((event) => {	2✔
58	this.subscriber.notifications.removeAllListeners(event)	6✔
59	})	2✔
60	await this.subscriber.close()	2✔
61	logSchema.info(logger, '[PubSub] Exited', {	2✔
62	type: 'pubsub',	2✔
63	})	2✔
64	}	2✔
65		1✔
66	async publish(channel: string, payload: unknown): Promise<void> {	1✔
67	await this.subscriber.notify(channel, payload)	×
UNCOV 68	}	×
69		1✔
70	async subscribe(channel: string, cb: (payload: any) => void): Promise<void> {	1✔
71	const listenerCount = this.subscriber.notifications.listenerCount(channel)	266✔
72	this.subscriber.notifications.on(channel, cb)	266✔
73		266✔
74	if (this.isConnected && listenerCount === 0) {	266✔
75	await this.subscriber.listenTo(channel)	6✔
76	}	6✔
77	}	266✔
78		1✔
79	async unsubscribe(channel: string, cb: (payload: any) => void): Promise<void> {	1✔
80	this.subscriber.notifications.removeListener(channel, cb)	260✔
81		260✔
82	const isListening = this.subscriber.notifications.listenerCount(channel) > 0	260✔
83		260✔
84	if (!isListening && this.isConnected) {	260!
85	await this.subscriber.unlisten(channel)	×
UNCOV 86	}	×
87	}	260✔
88	}	1✔

supabase / storage / 24474844911

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous