30832958929

Committed 03 Aug 2026 04:36PM UTC coverage: 53.989% (-26.5%) from 80.494%

Build # 30832958929

Build Type

Pull #1292

github

Committed by

web-flow

Commit Message

Merge 477971401 into 42e86a3a7

Pull Request Pull Request #1292: feat: new queing system

Coverage Stats

3711 of 7412 branches covered (50.07%)

Branch coverage included in aggregate %.

140 of 337 new or added lines in 38 files covered. (41.54%)

3365 existing lines in 150 files now uncovered.

6941 of 12318 relevant lines covered (56.35%)

77.58 hits per line

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

70.31

/src/http/plugins/vector.ts

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

supabase / storage / 30832958929

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