31193672639

Committed 07 Aug 2026 03:38PM UTC coverage: 80.689% (+0.09%) from 80.602%

Build # 31193672639

Build Type

Pull #1309

github

Committed by

web-flow

Commit Message

Merge fda2cc23a into 6b6903971

Pull Request Pull Request #1309: fix: extract postgres commons for watt

Coverage Stats

5676 of 7564 branches covered (75.04%)

Branch coverage included in aggregate %.

104 of 111 new or added lines in 9 files covered. (93.69%)

10716 of 12751 relevant lines covered (84.04%)

512.18 hits per line

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

72.31

/src/http/plugins/vector.ts

import {
  attachPoolErrorHandler,
  getTenantConfig,
  multitenantPgExecutor,
  PgPoolExecutor,
} from '@internal/database'
import { deriveVectorDatabaseUrl } from '@internal/database/vector-store-url'
import { ERRORS } from '@internal/errors'
import { logger, logSchema } from '@internal/monitoring'
import {
  BucketScopedSingleShard,
  PgShardStoreFactory,
  ShardCatalog,
  Sharder,
  SingleShard,
} from '@internal/sharding'
import {
  createS3VectorClient,
  createVectorTransactionPgResolver,
  PgVectorMetadataDB,
  PgVectorStore,
  S3Vector,
  VectorStore,
  VectorStoreManager,
} from '@storage/protocols/vector'
import { FastifyInstance } from 'fastify'
import fastifyPlugin from 'fastify-plugin'
import { Pool as PgPool } from 'pg'
import { getConfig } from '../../config'

declare module 'fastify' {
  interface FastifyRequest {
    s3Vector: VectorStoreManager
  }
}

export const s3vector = fastifyPlugin(async function (fastify: FastifyInstance) {
  const config = getConfig()
  const {
    vectorBucketProvider,
    vectorDatabaseCreate,
    vectorDatabaseURL,
    vectorS3Buckets,
    isMultitenant,
    databaseApplicationName,
  } = config

  let s3Adapter: S3Vector | undefined
  if (vectorBucketProvider === 's3' && vectorS3Buckets.length > 0) {
    s3Adapter = new S3Vector(createS3VectorClient())
  }

  // pgvector + single-tenant: VECTOR_DATABASE_URL is the maintenance URL the
  // migration runner used to CREATE DATABASE; the runtime pool targets the
  // derived `storage_vectors` database on the same server. When
  // VECTOR_DATABASE_CREATE=false, the runtime pool targets VECTOR_DATABASE_URL
  // directly.
  let stPgVectorAdapter: PgVectorStore | undefined
  let stPgVectorPool: PgPool | undefined
  if (vectorBucketProvider === 'pgvector' && !isMultitenant && vectorDatabaseURL) {
    const connectionString = vectorDatabaseCreate
      ? deriveVectorDatabaseUrl(vectorDatabaseURL)
      : vectorDatabaseURL
    stPgVectorPool = attachPoolErrorHandler(
      new PgPool({
        connectionString,
        application_name: databaseApplicationName,
        min: 0,
        max: 10,
      }),
      (error) => {
        logSchema.warning(logger, '[Vector] Idle pgvector client error', {
          type: 'db',
          error,
        })
      }
    )
    stPgVectorAdapter = new PgVectorStore(new PgPoolExecutor(stPgVectorPool))
    fastify.addHook('onClose', async () => {
      await stPgVectorPool?.end()
    })
  }

  const featureEnabled =
    (vectorBucketProvider === 's3' && Boolean(s3Adapter)) ||
    (vectorBucketProvider === 'pgvector' && (isMultitenant || Boolean(stPgVectorAdapter)))

  fastify.addHook('preHandler', async (req) => {
    if (!featureEnabled) {
      throw ERRORS.FeatureNotEnabled('vector', 'Vector service not configured')
    }

    const { vectorMaxBucketsCount, vectorMaxIndexesCount } = config

    let maxBucketCount = vectorMaxBucketsCount
    let maxIndexCount = vectorMaxIndexesCount

    if (isMultitenant) {
      const { features } = await getTenantConfig(req.tenantId)
      maxBucketCount = features?.vectorBuckets?.maxBuckets || vectorMaxBucketsCount
      maxIndexCount = features?.vectorBuckets?.maxIndexes || vectorMaxIndexesCount
    }

    const store = new PgVectorMetadataDB(req.db)

    let adapter: VectorStore
    if (vectorBucketProvider === 'pgvector') {
      adapter = isMultitenant
        ? new PgVectorStore(createVectorTransactionPgResolver(req.db))
        : stPgVectorAdapter!
    } else {
      adapter = s3Adapter!
    }

    let shard: Sharder
    if (vectorBucketProvider === 'pgvector') {
      shard = new BucketScopedSingleShard({
        keyPrefix: 'pgvector__',
        capacity: Number.MAX_SAFE_INTEGER,
      })
    } else if (isMultitenant) {
      shard = new ShardCatalog(new PgShardStoreFactory(multitenantPgExecutor))
    } else {
      shard = new SingleShard({
        shardKey: vectorS3Buckets[0],
        capacity: 10000,
      })
    }

    req.s3Vector = new VectorStoreManager(adapter, store, shard, {
      tenantId: req.tenantId,
      maxBucketCount,
      maxIndexCount,
    })
  })
})

1	import {
2	attachPoolErrorHandler,
3	getTenantConfig,
4	multitenantPgExecutor,
5	PgPoolExecutor,
6	} from '@internal/database'
7	import { deriveVectorDatabaseUrl } from '@internal/database/vector-store-url'
8	import { ERRORS } from '@internal/errors'
9	import { logger, logSchema } from '@internal/monitoring'
10	import {
11	BucketScopedSingleShard,
12	PgShardStoreFactory,
13	ShardCatalog,
14	Sharder,
15	SingleShard,
16	} from '@internal/sharding'
17	import {
18	createS3VectorClient,
19	createVectorTransactionPgResolver,
20	PgVectorMetadataDB,
21	PgVectorStore,
22	S3Vector,
23	VectorStore,
24	VectorStoreManager,
25	} from '@storage/protocols/vector'
26	import { FastifyInstance } from 'fastify'
27	import fastifyPlugin from 'fastify-plugin'
28	import { Pool as PgPool } from 'pg'
29	import { getConfig } from '../../config'
30
31	declare module 'fastify' {
32	interface FastifyRequest {
33	s3Vector: VectorStoreManager
34	}
35	}
36
37	export const s3vector = fastifyPlugin(async function (fastify: FastifyInstance) {	42✔
38	const config = getConfig()	295✔
39	const {
40	vectorBucketProvider,
41	vectorDatabaseCreate,
42	vectorDatabaseURL,
43	vectorS3Buckets,
44	isMultitenant,
45	databaseApplicationName,
46	} = config	295✔
47
48	let s3Adapter: S3Vector \| undefined
49	if (vectorBucketProvider === 's3' && vectorS3Buckets.length > 0) {	295✔
50	s3Adapter = new S3Vector(createS3VectorClient())	293✔
51	}
52
53	// pgvector + single-tenant: VECTOR_DATABASE_URL is the maintenance URL the
54	// migration runner used to CREATE DATABASE; the runtime pool targets the
55	// derived `storage_vectors` database on the same server. When
56	// VECTOR_DATABASE_CREATE=false, the runtime pool targets VECTOR_DATABASE_URL
57	// directly.
58	let stPgVectorAdapter: PgVectorStore \| undefined
59	let stPgVectorPool: PgPool \| undefined
60	if (vectorBucketProvider === 'pgvector' && !isMultitenant && vectorDatabaseURL) {	295✔
61	const connectionString = vectorDatabaseCreate	1!
62	? deriveVectorDatabaseUrl(vectorDatabaseURL)
63	: vectorDatabaseURL
64	stPgVectorPool = attachPoolErrorHandler(	1✔
65	new PgPool({
66	connectionString,
67	application_name: databaseApplicationName,
68	min: 0,
69	max: 10,
70	}),
71	(error) => {
NEW 72	logSchema.warning(logger, '[Vector] Idle pgvector client error', {	×
73	type: 'db',
74	error,
75	})
76	}
77	)
78	stPgVectorAdapter = new PgVectorStore(new PgPoolExecutor(stPgVectorPool))	1✔
79	fastify.addHook('onClose', async () => {	1✔
80	await stPgVectorPool?.end()	1✔
81	})
82	}
83
84	const featureEnabled =
85	(vectorBucketProvider === 's3' && Boolean(s3Adapter)) \|\|	295✔
86	(vectorBucketProvider === 'pgvector' && (isMultitenant \|\| Boolean(stPgVectorAdapter)))
87
88	fastify.addHook('preHandler', async (req) => {	295✔
89	if (!featureEnabled) {	152✔
90	throw ERRORS.FeatureNotEnabled('vector', 'Vector service not configured')	1✔
91	}
92
93	const { vectorMaxBucketsCount, vectorMaxIndexesCount } = config	151✔
94
95	let maxBucketCount = vectorMaxBucketsCount	151✔
96	let maxIndexCount = vectorMaxIndexesCount	151✔
97
98	if (isMultitenant) {	151!
99	const { features } = await getTenantConfig(req.tenantId)	×
100	maxBucketCount = features?.vectorBuckets?.maxBuckets \|\| vectorMaxBucketsCount	×
101	maxIndexCount = features?.vectorBuckets?.maxIndexes \|\| vectorMaxIndexesCount	×
102	}
103
104	const store = new PgVectorMetadataDB(req.db)	151✔
105
106	let adapter: VectorStore
107	if (vectorBucketProvider === 'pgvector') {	151!
108	adapter = isMultitenant	×
109	? new PgVectorStore(createVectorTransactionPgResolver(req.db))
110	: stPgVectorAdapter!
111	} else {
112	adapter = s3Adapter!	151✔
113	}
114
115	let shard: Sharder
116	if (vectorBucketProvider === 'pgvector') {	151!
117	shard = new BucketScopedSingleShard({	×
118	keyPrefix: 'pgvector__',
119	capacity: Number.MAX_SAFE_INTEGER,
120	})
121	} else if (isMultitenant) {	151!
122	shard = new ShardCatalog(new PgShardStoreFactory(multitenantPgExecutor))	×
123	} else {
124	shard = new SingleShard({	151✔
125	shardKey: vectorS3Buckets[0],
126	capacity: 10000,
127	})
128	}
129
130	req.s3Vector = new VectorStoreManager(adapter, store, shard, {	151✔
131	tenantId: req.tenantId,
132	maxBucketCount,
133	maxIndexCount,
134	})
135	})
136	})

supabase / storage / 31193672639

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