25738155938

Committed 12 May 2026 01:38PM UTC coverage: 39.251% (-35.1%) from 74.366%

Build # 25738155938

Build Type

Pull #1094

github

Committed by

web-flow

Commit Message

Merge 0f3efcca0 into defbbb616

Pull Request Pull Request #1094: feat: embedded vector store

Coverage Stats

2188 of 6152 branches covered (35.57%)

Branch coverage included in aggregate %.

88 of 280 new or added lines in 6 files covered. (31.43%)

3689 existing lines in 165 files now uncovered.

4312 of 10408 relevant lines covered (41.43%)

34.74 hits per line

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

6.73

/src/http/routes/admin/objects.ts

import { render } from '@internal/errors'
import { logSchema } from '@internal/monitoring'
import { ObjectScanner } from '@storage/scanner/scanner'
import { FastifyInstance, RequestGenericInterface } from 'fastify'
import { FastifyReply } from 'fastify/types/reply'
import { dbSuperUser, storage } from '../../plugins'
import { registerApiKeyAuth } from '../../plugins/apikey'

const listOrphanedObjects = {
  description: 'List Orphaned Objects',
  params: {
    type: 'object',
    properties: {
      tenantId: { type: 'string' },
      bucketId: { type: 'string' },
    },
    required: ['tenantId', 'bucketId'],
  },
  query: {
    type: 'object',
    properties: {
      before: { type: 'string' },
      keepTmpTable: { type: 'boolean' },
    },
  },
} as const

const syncOrphanedObjects = {
  description: 'Sync Orphaned Objects',
  params: {
    type: 'object',
    properties: {
      tenantId: { type: 'string' },
      bucketId: { type: 'string' },
    },
    required: ['tenantId', 'bucketId'],
  },
  body: {
    type: 'object',
    properties: {
      deleteDbKeys: { type: 'boolean' },
      deleteS3Keys: { type: 'boolean' },
      tmpTable: { type: 'string' },
    },
  },
  optional: ['deleteDbKeys', 'deleteS3Keys'],
} as const

interface ListOrphanObjectsRequest extends RequestGenericInterface {
  Params: {
    tenantId: string
    bucketId: string
  }
  Querystring: {
    before?: string
    keepTmpTable?: boolean
  }
}

interface SyncOrphanObjectsRequest extends RequestGenericInterface {
  Params: {
    tenantId: string
    bucketId: string
  }
  Body: {
    deleteDbKeys?: boolean
    deleteS3Keys?: boolean
    before?: string
    tmpTable?: string
    keepTmpTable?: boolean
  }
}

export default async function routes(fastify: FastifyInstance) {
  registerApiKeyAuth(fastify)
  fastify.register(dbSuperUser, {
    disableHostCheck: true,
    maxConnections: 5,
  })
  fastify.register(storage)

  fastify.get<ListOrphanObjectsRequest>(
    '/:tenantId/buckets/:bucketId/orphan-objects',
    {
      schema: { ...listOrphanedObjects, tags: ['object'] },
    },
    async (req, reply) => {
      const bucket = req.params.bucketId
      let before = req.query.before ? new Date(req.query.before as string) : undefined

      if (before && isNaN(before.getTime())) {
        return reply.status(400).send({
          error: 'Invalid date format',
        })
      }
      if (!before) {
        before = new Date()
        before.setHours(before.getHours() - 1)
      }

      const scanner = new ObjectScanner(req.storage)
      const orphanObjects = scanner.listOrphaned(bucket, {
        signal: req.signals.disconnect.signal,
        before,
        keepTmpTable: Boolean(req.query.keepTmpTable),
      })

      reply.header('Content-Type', 'application/x-ndjson; charset=utf-8')

      // Do not let the connection time out, periodically send
      // a ping message to keep the connection alive
      const respPing = ping(reply)

      try {
        for await (const result of orphanObjects) {
          if (result.value.length > 0) {
            respPing.update()
            writeNdjson(reply, {
              ...result,
              event: 'data',
            })
          }
        }
      } catch (e) {
        logSchema.error(req.log, 'list orphaned objects stream failed', {
          type: 'orphan',
          tenantId: req.tenantId,
          project: req.tenantId,
          reqId: req.id,
          sbReqId: req.sbReqId,
          error: e,
          metadata: JSON.stringify({ bucket }),
        })
        writeNdjson(reply, {
          event: 'error',
          error: render(e),
        })
        return
      } finally {
        respPing.clear()
        endNdjson(reply)
      }
    }
  )

  fastify.delete<SyncOrphanObjectsRequest>(
    '/:tenantId/buckets/:bucketId/orphan-objects',
    {
      schema: { ...syncOrphanedObjects, tags: ['object'] },
    },
    async (req, reply) => {
      if (!req.body.deleteDbKeys && !req.body.deleteS3Keys) {
        return reply.status(400).send({
          error: 'At least one of deleteDbKeys or deleteS3Keys must be set to true',
        })
      }

      const bucket = `${req.params.bucketId}`
      let before = req.body.before ? new Date(req.body.before as string) : undefined

      if (before && isNaN(before.getTime())) {
        return reply.status(400).send({
          error: 'Invalid date format',
        })
      }
      if (!before) {
        before = new Date()
        before.setHours(before.getHours() - 1)
      }

      reply.header('Content-Type', 'application/x-ndjson; charset=utf-8')

      const respPing = ping(reply)

      try {
        const scanner = new ObjectScanner(req.storage)
        const result = scanner.deleteOrphans(bucket, {
          deleteDbKeys: req.body.deleteDbKeys,
          deleteS3Keys: req.body.deleteS3Keys,
          signal: req.signals.disconnect.signal,
          before,
          tmpTable: req.body.tmpTable,
        })

        for await (const deleted of result) {
          respPing.update()
          writeNdjson(reply, {
            ...deleted,
            event: 'data',
          })
        }
      } catch (e) {
        logSchema.error(req.log, 'delete orphaned objects stream failed', {
          type: 'orphan',
          tenantId: req.tenantId,
          project: req.tenantId,
          reqId: req.id,
          sbReqId: req.sbReqId,
          error: e,
          metadata: JSON.stringify({ bucket }),
        })
        writeNdjson(reply, {
          event: 'error',
          error: render(e),
        })
        return
      } finally {
        respPing.clear()
        endNdjson(reply)
      }
    }
  )
}

function canWriteNdjson(reply: FastifyReply) {
  return !reply.raw.destroyed && !reply.raw.writableEnded
}

function writeNdjson(reply: FastifyReply, payload: unknown) {
  if (!canWriteNdjson(reply)) {
    return false
  }

  try {
    reply.raw.write(JSON.stringify(payload) + '\n')
    return true
  } catch {
    return false
  }
}

function endNdjson(reply: FastifyReply) {
  if (!canWriteNdjson(reply)) {
    return
  }

  try {
    reply.raw.end()
  } catch {}
}

// Occasionally write a ping message to the response stream
function ping(reply: FastifyReply) {
  let lastSend = undefined as Date | undefined
  const clearPing = setInterval(() => {
    const fiveSecondsEarly = new Date()
    fiveSecondsEarly.setSeconds(fiveSecondsEarly.getSeconds() - 5)

    if (!lastSend || (lastSend && lastSend < fiveSecondsEarly)) {
      lastSend = new Date()
      writeNdjson(reply, {
        event: 'ping',
      })
    }
  }, 1000 * 10)

  return {
    clear: () => clearInterval(clearPing),
    update: () => {
      lastSend = new Date()
    },
  }
}

1	import { render } from '@internal/errors'
2	import { logSchema } from '@internal/monitoring'
3	import { ObjectScanner } from '@storage/scanner/scanner'
4	import { FastifyInstance, RequestGenericInterface } from 'fastify'
5	import { FastifyReply } from 'fastify/types/reply'
6	import { dbSuperUser, storage } from '../../plugins'
7	import { registerApiKeyAuth } from '../../plugins/apikey'
8
9	const listOrphanedObjects = {	2✔
10	description: 'List Orphaned Objects',
11	params: {
12	type: 'object',
13	properties: {
14	tenantId: { type: 'string' },
15	bucketId: { type: 'string' },
16	},
17	required: ['tenantId', 'bucketId'],
18	},
19	query: {
20	type: 'object',
21	properties: {
22	before: { type: 'string' },
23	keepTmpTable: { type: 'boolean' },
24	},
25	},
26	} as const
27
28	const syncOrphanedObjects = {	2✔
29	description: 'Sync Orphaned Objects',
30	params: {
31	type: 'object',
32	properties: {
33	tenantId: { type: 'string' },
34	bucketId: { type: 'string' },
35	},
36	required: ['tenantId', 'bucketId'],
37	},
38	body: {
39	type: 'object',
40	properties: {
41	deleteDbKeys: { type: 'boolean' },
42	deleteS3Keys: { type: 'boolean' },
43	tmpTable: { type: 'string' },
44	},
45	},
46	optional: ['deleteDbKeys', 'deleteS3Keys'],
47	} as const
48
49	interface ListOrphanObjectsRequest extends RequestGenericInterface {
50	Params: {
51	tenantId: string
52	bucketId: string
53	}
54	Querystring: {
55	before?: string
56	keepTmpTable?: boolean
57	}
58	}
59
60	interface SyncOrphanObjectsRequest extends RequestGenericInterface {
61	Params: {
62	tenantId: string
63	bucketId: string
64	}
65	Body: {
66	deleteDbKeys?: boolean
67	deleteS3Keys?: boolean
68	before?: string
69	tmpTable?: string
70	keepTmpTable?: boolean
71	}
72	}
73
74	export default async function routes(fastify: FastifyInstance) {
75	registerApiKeyAuth(fastify)	2✔
76	fastify.register(dbSuperUser, {	2✔
77	disableHostCheck: true,
78	maxConnections: 5,
79	})
80	fastify.register(storage)	2✔
81
82	fastify.get<ListOrphanObjectsRequest>(	2✔
83	'/:tenantId/buckets/:bucketId/orphan-objects',
84	{
85	schema: { ...listOrphanedObjects, tags: ['object'] },
86	},
87	async (req, reply) => {
UNCOV 88	const bucket = req.params.bucketId	×
UNCOV 89	let before = req.query.before ? new Date(req.query.before as string) : undefined	×
90
UNCOV 91	if (before && isNaN(before.getTime())) {	×
UNCOV 92	return reply.status(400).send({	×
93	error: 'Invalid date format',
94	})
95	}
96	if (!before) {	×
97	before = new Date()	×
98	before.setHours(before.getHours() - 1)	×
99	}
100
101	const scanner = new ObjectScanner(req.storage)	×
102	const orphanObjects = scanner.listOrphaned(bucket, {	×
103	signal: req.signals.disconnect.signal,
104	before,
105	keepTmpTable: Boolean(req.query.keepTmpTable),
106	})
107
108	reply.header('Content-Type', 'application/x-ndjson; charset=utf-8')	×
109
110	// Do not let the connection time out, periodically send
111	// a ping message to keep the connection alive
112	const respPing = ping(reply)	×
113
114	try {	×
115	for await (const result of orphanObjects) {	×
116	if (result.value.length > 0) {	×
117	respPing.update()	×
118	writeNdjson(reply, {	×
119	...result,
120	event: 'data',
121	})
122	}
123	}
124	} catch (e) {
125	logSchema.error(req.log, 'list orphaned objects stream failed', {	×
126	type: 'orphan',
127	tenantId: req.tenantId,
128	project: req.tenantId,
129	reqId: req.id,
130	sbReqId: req.sbReqId,
131	error: e,
132	metadata: JSON.stringify({ bucket }),
133	})
134	writeNdjson(reply, {	×
135	event: 'error',
136	error: render(e),
137	})
138	return	×
139	} finally {
140	respPing.clear()	×
141	endNdjson(reply)	×
142	}
143	}
144	)
145
146	fastify.delete<SyncOrphanObjectsRequest>(	2✔
147	'/:tenantId/buckets/:bucketId/orphan-objects',
148	{
149	schema: { ...syncOrphanedObjects, tags: ['object'] },
150	},
151	async (req, reply) => {
UNCOV 152	if (!req.body.deleteDbKeys && !req.body.deleteS3Keys) {	×
UNCOV 153	return reply.status(400).send({	×
154	error: 'At least one of deleteDbKeys or deleteS3Keys must be set to true',
155	})
156	}
157
UNCOV 158	const bucket = `${req.params.bucketId}`	×
UNCOV 159	let before = req.body.before ? new Date(req.body.before as string) : undefined	×
160
UNCOV 161	if (before && isNaN(before.getTime())) {	×
UNCOV 162	return reply.status(400).send({	×
163	error: 'Invalid date format',
164	})
165	}
166	if (!before) {	×
167	before = new Date()	×
168	before.setHours(before.getHours() - 1)	×
169	}
170
171	reply.header('Content-Type', 'application/x-ndjson; charset=utf-8')	×
172
173	const respPing = ping(reply)	×
174
175	try {	×
176	const scanner = new ObjectScanner(req.storage)	×
177	const result = scanner.deleteOrphans(bucket, {	×
178	deleteDbKeys: req.body.deleteDbKeys,
179	deleteS3Keys: req.body.deleteS3Keys,
180	signal: req.signals.disconnect.signal,
181	before,
182	tmpTable: req.body.tmpTable,
183	})
184
185	for await (const deleted of result) {	×
186	respPing.update()	×
187	writeNdjson(reply, {	×
188	...deleted,
189	event: 'data',
190	})
191	}
192	} catch (e) {
193	logSchema.error(req.log, 'delete orphaned objects stream failed', {	×
194	type: 'orphan',
195	tenantId: req.tenantId,
196	project: req.tenantId,
197	reqId: req.id,
198	sbReqId: req.sbReqId,
199	error: e,
200	metadata: JSON.stringify({ bucket }),
201	})
202	writeNdjson(reply, {	×
203	event: 'error',
204	error: render(e),
205	})
206	return	×
207	} finally {
208	respPing.clear()	×
209	endNdjson(reply)	×
210	}
211	}
212	)
213	}
214
215	function canWriteNdjson(reply: FastifyReply) {
216	return !reply.raw.destroyed && !reply.raw.writableEnded	×
217	}
218
219	function writeNdjson(reply: FastifyReply, payload: unknown) {
220	if (!canWriteNdjson(reply)) {	×
221	return false	×
222	}
223
224	try {	×
225	reply.raw.write(JSON.stringify(payload) + '\n')	×
226	return true	×
227	} catch {
228	return false	×
229	}
230	}
231
232	function endNdjson(reply: FastifyReply) {
233	if (!canWriteNdjson(reply)) {	×
234	return	×
235	}
236
237	try {	×
238	reply.raw.end()	×
239	} catch {}
240	}
241
242	// Occasionally write a ping message to the response stream
243	function ping(reply: FastifyReply) {
244	let lastSend = undefined as Date \| undefined	×
245	const clearPing = setInterval(() => {	×
246	const fiveSecondsEarly = new Date()	×
247	fiveSecondsEarly.setSeconds(fiveSecondsEarly.getSeconds() - 5)	×
248
249	if (!lastSend \|\| (lastSend && lastSend < fiveSecondsEarly)) {	×
250	lastSend = new Date()	×
251	writeNdjson(reply, {	×
252	event: 'ping',
253	})
254	}
255	}, 1000 * 10)
256
257	return {	×
258	clear: () => clearInterval(clearPing),	×
259	update: () => {
260	lastSend = new Date()	×
261	},
262	}
263	}

supabase / storage / 25738155938

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