12743727259

Committed 13 Jan 2025 08:57AM UTC coverage: 77.349% (-0.7%) from 78.085%

Build # 12743727259

Build Type

Pull #606

github

Committed by

web-flow

Commit Message

Merge 78ab89e1c into b54c39518

Pull Request Pull Request #606: feat: reconcile orphan objects from admin endpoint

Coverage Stats

1272 of 1800 branches covered (70.67%)

Branch coverage included in aggregate %.

912 of 1344 new or added lines in 25 files covered. (67.86%)

14 existing lines in 2 files now uncovered.

15119 of 19391 relevant lines covered (77.97%)

155.82 hits per line

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

45.93

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

import { FastifyInstance, RequestGenericInterface } from 'fastify'
import apiKey from '../../plugins/apikey'
import { dbSuperUser, storage } from '../../plugins'
import { ObjectScanner } from '@storage/scanner/scanner'
import { FastifyReply } from 'fastify/types/reply'
import { getConfig } from '../../../config'

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' },
    },
  },
} 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' },
    },
  },
  optional: ['deleteDbKeys', 'deleteS3Keys'],
} as const

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

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

const { storageS3BackupBucket } = getConfig()

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

  fastify.get<ListOrphanObjectsRequest>(
    '/:tenantId/buckets/:bucketId/orphan-objects',
    {
      schema: listOrphanedObjects,
    },
    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: before,
      })

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

      // Do not 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()
            reply.raw.write(
              JSON.stringify({
                ...result,
                type: 'data',
              })
            )
          }
        }
      } catch (e) {
        throw e
      } finally {
        respPing.clear()
        reply.raw.end()
      }
    }
  )

  fastify.delete<SyncOrphanObjectsRequest>(
    '/:tenantId/buckets/:bucketId/orphan-objects',
    {
      schema: syncOrphanedObjects,
    },
    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',
        })
      }

      if (!storageS3BackupBucket) {
        return reply.status(400).send({
          error: 'Backup bucket not configured',
        })
      }

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

      if (!before) {
        before = new Date()
        before.setHours(before.getHours() - 1)
      }

      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,
        })

        for await (const deleted of result) {
          respPing.update()
          reply.raw.write(
            JSON.stringify({
              ...deleted,
              type: 'data',
            })
          )
        }
      } catch (e) {
        throw e
      } finally {
        respPing.clear()
        reply.raw.end()
      }
    }
  )
}

// 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()
      reply.raw.write(
        JSON.stringify({
          type: 'ping',
        })
      )
    }
  }, 1000 * 10)

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

1	import { FastifyInstance, RequestGenericInterface } from 'fastify'	1✔
2	import apiKey from '../../plugins/apikey'	1✔
3	import { dbSuperUser, storage } from '../../plugins'	1✔
4	import { ObjectScanner } from '@storage/scanner/scanner'	1✔
5	import { FastifyReply } from 'fastify/types/reply'	1✔
6	import { getConfig } from '../../../config'	1✔
7		1✔
8	const listOrphanedObjects = {	1✔
9	description: 'List Orphaned Objects',	1✔
10	params: {	1✔
11	type: 'object',	1✔
12	properties: {	1✔
13	tenantId: { type: 'string' },	1✔
14	bucketId: { type: 'string' },	1✔
15	},	1✔
16	required: ['tenantId', 'bucketId'],	1✔
17	},	1✔
18	query: {	1✔
19	type: 'object',	1✔
20	properties: {	1✔
21	before: { type: 'string' },	1✔
22	},	1✔
23	},	1✔
24	} as const	1✔
25		1✔
26	const syncOrphanedObjects = {	1✔
27	description: 'Sync Orphaned Objects',	1✔
28	params: {	1✔
29	type: 'object',	1✔
30	properties: {	1✔
31	tenantId: { type: 'string' },	1✔
32	bucketId: { type: 'string' },	1✔
33	},	1✔
34	required: ['tenantId', 'bucketId'],	1✔
35	},	1✔
36	body: {	1✔
37	type: 'object',	1✔
38	properties: {	1✔
39	deleteDbKeys: { type: 'boolean' },	1✔
40	deleteS3Keys: { type: 'boolean' },	1✔
41	},	1✔
42	},	1✔
43	optional: ['deleteDbKeys', 'deleteS3Keys'],	1✔
44	} as const	1✔
45		1✔
46	interface ListOrphanObjectsRequest extends RequestGenericInterface {	1✔
47	Params: {	1✔
48	tenantId: string	1✔
49	bucketId: string	1✔
50	}	1✔
51	Querystring: {	1✔
52	before?: string	1✔
53	}	1✔
54	}	1✔
55		1✔
56	interface SyncOrphanObjectsRequest extends RequestGenericInterface {	1✔
57	Params: {	1✔
58	tenantId: string	1✔
59	bucketId: string	1✔
60	}	1✔
61	Body: {	1✔
62	deleteDbKeys?: boolean	1✔
63	deleteS3Keys?: boolean	1✔
64	before?: string	1✔
65	}	1✔
66	}	1✔
67		1✔
68	const { storageS3BackupBucket } = getConfig()	1✔
69		1✔
70	export default async function routes(fastify: FastifyInstance) {	1✔
71	fastify.register(apiKey)	1✔
72	fastify.register(dbSuperUser, {	1✔
73	disableHostCheck: true,	1✔
74	maxConnections: 5,	1✔
75	})	1✔
76	fastify.register(storage)	1✔
77		1✔
78	fastify.get<ListOrphanObjectsRequest>(	1✔
79	'/:tenantId/buckets/:bucketId/orphan-objects',	1✔
80	{	1✔
81	schema: listOrphanedObjects,	1✔
82	},	1✔
83	async (req, reply) => {	1✔
NEW 84	const bucket = req.params.bucketId	×
NEW 85	let before = req.query.before ? new Date(req.query.before as string) : undefined	×
NEW 86		×
NEW 87	if (before && isNaN(before.getTime())) {	×
NEW 88	return reply.status(400).send({	×
NEW 89	error: 'Invalid date format',	×
NEW 90	})	×
NEW 91	}	×
NEW 92	if (!before) {	×
NEW 93	before = new Date()	×
NEW 94	before.setHours(before.getHours() - 1)	×
NEW 95	}	×
NEW 96		×
NEW 97	const scanner = new ObjectScanner(req.storage)	×
NEW 98	const orphanObjects = scanner.listOrphaned(bucket, {	×
NEW 99	signal: req.signals.disconnect.signal,	×
NEW 100	before: before,	×
NEW 101	})	×
NEW 102		×
NEW 103	reply.header('Content-Type', 'application/json; charset=utf-8')	×
NEW 104		×
NEW 105	// Do not the connection time out, periodically send	×
NEW 106	// a ping message to keep the connection alive	×
NEW 107	const respPing = ping(reply)	×
NEW 108		×
NEW 109	try {	×
NEW 110	for await (const result of orphanObjects) {	×
NEW 111	if (result.value.length > 0) {	×
NEW 112	respPing.update()	×
NEW 113	reply.raw.write(	×
NEW 114	JSON.stringify({	×
NEW 115	...result,	×
NEW 116	type: 'data',	×
NEW 117	})	×
NEW 118	)	×
NEW 119	}	×
NEW 120	}	×
NEW 121	} catch (e) {	×
NEW 122	throw e	×
NEW 123	} finally {	×
NEW 124	respPing.clear()	×
NEW 125	reply.raw.end()	×
NEW 126	}	×
NEW 127	}	×
128	)	1✔
129		1✔
130	fastify.delete<SyncOrphanObjectsRequest>(	1✔
131	'/:tenantId/buckets/:bucketId/orphan-objects',	1✔
132	{	1✔
133	schema: syncOrphanedObjects,	1✔
134	},	1✔
135	async (req, reply) => {	1✔
NEW 136	if (!req.body.deleteDbKeys && !req.body.deleteS3Keys) {	×
NEW 137	return reply.status(400).send({	×
NEW 138	error: 'At least one of deleteDbKeys or deleteS3Keys must be set to true',	×
NEW 139	})	×
NEW 140	}	×
NEW 141		×
NEW 142	if (!storageS3BackupBucket) {	×
NEW 143	return reply.status(400).send({	×
NEW 144	error: 'Backup bucket not configured',	×
NEW 145	})	×
NEW 146	}	×
NEW 147		×
NEW 148	const bucket = `${req.params.bucketId}`	×
NEW 149	let before = req.body.before ? new Date(req.body.before as string) : undefined	×
NEW 150		×
NEW 151	if (!before) {	×
NEW 152	before = new Date()	×
NEW 153	before.setHours(before.getHours() - 1)	×
NEW 154	}	×
NEW 155		×
NEW 156	const respPing = ping(reply)	×
NEW 157		×
NEW 158	try {	×
NEW 159	const scanner = new ObjectScanner(req.storage)	×
NEW 160	const result = scanner.deleteOrphans(bucket, {	×
NEW 161	deleteDbKeys: req.body.deleteDbKeys,	×
NEW 162	deleteS3Keys: req.body.deleteS3Keys,	×
NEW 163	signal: req.signals.disconnect.signal,	×
NEW 164	})	×
NEW 165		×
NEW 166	for await (const deleted of result) {	×
NEW 167	respPing.update()	×
NEW 168	reply.raw.write(	×
NEW 169	JSON.stringify({	×
NEW 170	...deleted,	×
NEW 171	type: 'data',	×
NEW 172	})	×
NEW 173	)	×
NEW 174	}	×
NEW 175	} catch (e) {	×
NEW 176	throw e	×
NEW 177	} finally {	×
NEW 178	respPing.clear()	×
NEW 179	reply.raw.end()	×
NEW 180	}	×
NEW 181	}	×
182	)	1✔
183	}	1✔
184		1✔
185	// Occasionally write a ping message to the response stream	1✔
NEW 186	function ping(reply: FastifyReply) {	×
NEW 187	let lastSend = undefined as Date \| undefined	×
NEW 188	const clearPing = setInterval(() => {	×
NEW 189	const fiveSecondsEarly = new Date()	×
NEW 190	fiveSecondsEarly.setSeconds(fiveSecondsEarly.getSeconds() - 5)	×
NEW 191		×
NEW 192	if (!lastSend \|\| (lastSend && lastSend < fiveSecondsEarly)) {	×
NEW 193	lastSend = new Date()	×
NEW 194	reply.raw.write(	×
NEW 195	JSON.stringify({	×
NEW 196	type: 'ping',	×
NEW 197	})	×
NEW 198	)	×
NEW 199	}	×
NEW 200	}, 1000 * 10)	×
NEW 201		×
NEW 202	return {	×
NEW 203	clear: () => clearInterval(clearPing),	×
NEW 204	update: () => {	×
NEW 205	lastSend = new Date()	×
NEW 206	},	×
NEW 207	}	×
NEW 208	}	×

supabase / storage / 12743727259

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