feat(media): index e delete em lote de mídias (Wasabi, coordenado DB↔storage)
- StorageProvider.deleteObjects(paths): delete seletivo em lote (chunks de 1000), no provider concreto (Wasabi DeleteObjectsCommand + fallback local). - GET /api/instances/:id/media-index: index/auditoria por chat/tipo/período (baseado no DB, com agregado byType, paginado). - POST /api/instances/:id/media-batch-delete: delete coordenado DB↔Wasabi, dryRun por padrão (precisa dryRun:false p/ apagar); limpa mediaPath/mediaUrl. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -103,6 +103,15 @@ class StorageProviderRegistry {
|
|||||||
return this.provider.deletePrefix(prefix)
|
return this.provider.deletePrefix(prefix)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async deleteObjects(paths: string[]): Promise<{ deleted: number }> {
|
||||||
|
if (!this.provider) {
|
||||||
|
rootLogger.warn('[StorageProvider] deleteObjects chamado mas nenhum provider está registrado')
|
||||||
|
return { deleted: 0 }
|
||||||
|
}
|
||||||
|
if (paths.length === 0) return { deleted: 0 }
|
||||||
|
return this.provider.deleteObjects(paths)
|
||||||
|
}
|
||||||
|
|
||||||
unregister(): void {
|
unregister(): void {
|
||||||
this.provider = null
|
this.provider = null
|
||||||
this.healthy = false
|
this.healthy = false
|
||||||
|
|||||||
@@ -133,6 +133,8 @@ export interface StorageProviderInterface {
|
|||||||
checkDetailedHealth(): Promise<import('./StorageProvider').StorageHealthStatus>
|
checkDetailedHealth(): Promise<import('./StorageProvider').StorageHealthStatus>
|
||||||
upload(payload: StorageUploadPayload): Promise<{ provider: string; path: string }>
|
upload(payload: StorageUploadPayload): Promise<{ provider: string; path: string }>
|
||||||
deletePrefix(prefix: string): Promise<{ deleted: number }>
|
deletePrefix(prefix: string): Promise<{ deleted: number }>
|
||||||
|
/** Deleta uma lista específica de paths (delete em lote seletivo, até 1000/req no S3) */
|
||||||
|
deleteObjects(paths: string[]): Promise<{ deleted: number }>
|
||||||
/** Recupera os bytes de um arquivo pelo seu path relativo no bucket */
|
/** Recupera os bytes de um arquivo pelo seu path relativo no bucket */
|
||||||
getBuffer(path: string): Promise<Buffer | null>
|
getBuffer(path: string): Promise<Buffer | null>
|
||||||
}
|
}
|
||||||
|
|||||||
+99
@@ -360,9 +360,108 @@ export function buildMediaRoutes(manager: WhatsAppConnectionManager): Router {
|
|||||||
}
|
}
|
||||||
})
|
})
|
||||||
|
|
||||||
|
// ── GET /api/instances/:instanceId/media-index ─────────────────────────────
|
||||||
|
// Index/auditoria de mídias desta instância (baseado no DB, sem listObjects).
|
||||||
|
// Query: chatId?, type?, from?, to? (ISO), page=1, pageSize=50
|
||||||
|
router.get('/media-index', async (req: Request, res: Response) => {
|
||||||
|
try {
|
||||||
|
const tenantId = req.tenantId!
|
||||||
|
const instanceId = req.params['instanceId'] as string
|
||||||
|
const where = buildMediaWhere(tenantId, instanceId, req.query)
|
||||||
|
|
||||||
|
const page = Math.max(1, parseInt(String(req.query['page'] ?? '1'), 10) || 1)
|
||||||
|
const pageSize = Math.min(200, Math.max(1, parseInt(String(req.query['pageSize'] ?? '50'), 10) || 50))
|
||||||
|
|
||||||
|
const [total, byTypeRaw, items] = await Promise.all([
|
||||||
|
prisma.message.count({ where }),
|
||||||
|
prisma.message.groupBy({ by: ['type'], where, _count: { _all: true } }),
|
||||||
|
prisma.message.findMany({
|
||||||
|
where,
|
||||||
|
select: { id: true, type: true, chatId: true, remoteJid: true, mediaPath: true, mediaUrl: true, mimeType: true, timestamp: true },
|
||||||
|
orderBy: { timestamp: 'desc' },
|
||||||
|
skip: (page - 1) * pageSize,
|
||||||
|
take: pageSize,
|
||||||
|
}),
|
||||||
|
])
|
||||||
|
|
||||||
|
const byType = Object.fromEntries(byTypeRaw.map((r) => [r.type, r._count._all]))
|
||||||
|
res.json({ total, page, pageSize, byType, items })
|
||||||
|
} catch (err: any) {
|
||||||
|
logger.error({ err }, 'Erro no media-index')
|
||||||
|
res.status(500).json({ error: err.message ?? 'Erro ao indexar mídias' })
|
||||||
|
}
|
||||||
|
})
|
||||||
|
|
||||||
|
// ── POST /api/instances/:instanceId/media-batch-delete ─────────────────────
|
||||||
|
// Delete em lote coordenado DB↔Wasabi. Body: { chatId?, type?, from?, to?, dryRun? }
|
||||||
|
// dryRun=true (padrão) só conta; para apagar de fato, enviar dryRun:false.
|
||||||
|
router.post('/media-batch-delete', async (req: Request, res: Response) => {
|
||||||
|
try {
|
||||||
|
const tenantId = req.tenantId!
|
||||||
|
const instanceId = req.params['instanceId'] as string
|
||||||
|
const { dryRun = true } = req.body ?? {}
|
||||||
|
const where = buildMediaWhere(tenantId, instanceId, req.body ?? {})
|
||||||
|
|
||||||
|
const matchedRows = await prisma.message.findMany({
|
||||||
|
where,
|
||||||
|
select: { id: true, mediaPath: true },
|
||||||
|
})
|
||||||
|
const matched = matchedRows.length
|
||||||
|
|
||||||
|
if (dryRun) {
|
||||||
|
res.json({ dryRun: true, matched, willDeleteObjects: matchedRows.filter((m) => m.mediaPath).length })
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if (matched === 0) {
|
||||||
|
res.json({ dryRun: false, matched: 0, deletedObjects: 0, dbCleared: 0 })
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
// 1. Apaga objetos do storage (só os que têm path relativo no bucket)
|
||||||
|
const paths = matchedRows.map((m) => m.mediaPath).filter((p): p is string => Boolean(p) && !p!.startsWith('http'))
|
||||||
|
let deletedObjects = 0
|
||||||
|
if (paths.length > 0 && storageProvider.isRegistered()) {
|
||||||
|
const r = await storageProvider.deleteObjects(paths)
|
||||||
|
deletedObjects = r.deleted
|
||||||
|
}
|
||||||
|
|
||||||
|
// 2. Limpa referências no DB (não deixa link quebrado nem órfão lógico)
|
||||||
|
const upd = await prisma.message.updateMany({
|
||||||
|
where: { id: { in: matchedRows.map((m) => m.id) } },
|
||||||
|
data: { mediaPath: null, mediaUrl: null },
|
||||||
|
})
|
||||||
|
|
||||||
|
logger.info({ instanceId, matched, deletedObjects, dbCleared: upd.count }, 'Delete em lote de mídias concluído')
|
||||||
|
res.json({ dryRun: false, matched, deletedObjects, dbCleared: upd.count })
|
||||||
|
} catch (err: any) {
|
||||||
|
logger.error({ err }, 'Erro no media-batch-delete')
|
||||||
|
res.status(500).json({ error: err.message ?? 'Erro ao apagar mídias em lote' })
|
||||||
|
}
|
||||||
|
})
|
||||||
|
|
||||||
return router
|
return router
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ─── Helper: monta o filtro Prisma para index/delete de mídias ────────────────
|
||||||
|
const MEDIA_TYPES = ['IMAGE', 'VIDEO', 'AUDIO', 'DOCUMENT', 'STICKER'] as const
|
||||||
|
|
||||||
|
function buildMediaWhere(tenantId: string, instanceId: string, src: Record<string, any>) {
|
||||||
|
const where: any = { tenantId, instanceId, mediaPath: { not: null } }
|
||||||
|
if (src['chatId']) where.chatId = String(src['chatId'])
|
||||||
|
if (src['type']) {
|
||||||
|
const t = String(src['type']).toUpperCase()
|
||||||
|
if ((MEDIA_TYPES as readonly string[]).includes(t)) where.type = t
|
||||||
|
}
|
||||||
|
const from = src['from'] ? new Date(String(src['from'])) : null
|
||||||
|
const to = src['to'] ? new Date(String(src['to'])) : null
|
||||||
|
if ((from && !isNaN(from.getTime())) || (to && !isNaN(to.getTime()))) {
|
||||||
|
where.timestamp = {}
|
||||||
|
if (from && !isNaN(from.getTime())) where.timestamp.gte = from
|
||||||
|
if (to && !isNaN(to.getTime())) where.timestamp.lte = to
|
||||||
|
}
|
||||||
|
return where
|
||||||
|
}
|
||||||
|
|
||||||
// ─── Helpers ──────────────────────────────────────────────────────────────────
|
// ─── Helpers ──────────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
const EXT_MAP: Record<string, string> = {
|
const EXT_MAP: Record<string, string> = {
|
||||||
|
|||||||
@@ -271,6 +271,42 @@ class UnifiedStorageProviderPlugin {
|
|||||||
}
|
}
|
||||||
return { deleted: deletedCount };
|
return { deleted: deletedCount };
|
||||||
}
|
}
|
||||||
|
/** Delete em lote seletivo: apaga exatamente os paths informados (não por prefixo). */
|
||||||
|
async deleteObjects(paths) {
|
||||||
|
const keys = paths.filter((p) => p && !p.startsWith('http'));
|
||||||
|
if (keys.length === 0)
|
||||||
|
return { deleted: 0 };
|
||||||
|
let deletedCount = 0;
|
||||||
|
// 1. Wasabi — em chunks de 1000 (limite do DeleteObjectsCommand)
|
||||||
|
if (this.s3 && this.bucket) {
|
||||||
|
for (let i = 0; i < keys.length; i += 1000) {
|
||||||
|
const chunk = keys.slice(i, i + 1000);
|
||||||
|
try {
|
||||||
|
await this.s3.send(new client_s3_1.DeleteObjectsCommand({
|
||||||
|
Bucket: this.bucket,
|
||||||
|
Delete: { Objects: chunk.map((Key) => ({ Key })), Quiet: true },
|
||||||
|
}));
|
||||||
|
deletedCount += chunk.length;
|
||||||
|
this.ctx.logger.info(`[UnifiedStorageProvider] ${chunk.length} objetos deletados do Wasabi (lote seletivo)`);
|
||||||
|
}
|
||||||
|
catch (err) {
|
||||||
|
this.ctx.logger.error(`[UnifiedStorageProvider] Erro no delete em lote do Wasabi: ${err.message}`);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
// 2. Fallback local
|
||||||
|
for (const key of keys) {
|
||||||
|
const localPath = path_1.default.join(this.localFallbackDir, key);
|
||||||
|
try {
|
||||||
|
if (fs_1.default.existsSync(localPath))
|
||||||
|
fs_1.default.unlinkSync(localPath);
|
||||||
|
}
|
||||||
|
catch (err) {
|
||||||
|
this.ctx.logger.error(`[UnifiedStorageProvider] Erro ao remover fallback local ${key}: ${err.message}`);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return { deleted: deletedCount };
|
||||||
|
}
|
||||||
async upload(payload) {
|
async upload(payload) {
|
||||||
const { category, file } = payload;
|
const { category, file } = payload;
|
||||||
if (!VALID_CATEGORIES.includes(category))
|
if (!VALID_CATEGORIES.includes(category))
|
||||||
|
|||||||
File diff suppressed because one or more lines are too long
@@ -301,6 +301,43 @@ class UnifiedStorageProviderPlugin implements PluginInstance, StorageProviderInt
|
|||||||
return { deleted: deletedCount };
|
return { deleted: deletedCount };
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/** Delete em lote seletivo: apaga exatamente os paths informados (não por prefixo). */
|
||||||
|
async deleteObjects(paths: string[]): Promise<{ deleted: number }> {
|
||||||
|
const keys = paths.filter((p) => p && !p.startsWith('http'));
|
||||||
|
if (keys.length === 0) return { deleted: 0 };
|
||||||
|
|
||||||
|
let deletedCount = 0;
|
||||||
|
|
||||||
|
// 1. Wasabi — em chunks de 1000 (limite do DeleteObjectsCommand)
|
||||||
|
if (this.s3 && this.bucket) {
|
||||||
|
for (let i = 0; i < keys.length; i += 1000) {
|
||||||
|
const chunk = keys.slice(i, i + 1000);
|
||||||
|
try {
|
||||||
|
await this.s3.send(new DeleteObjectsCommand({
|
||||||
|
Bucket: this.bucket,
|
||||||
|
Delete: { Objects: chunk.map((Key) => ({ Key })), Quiet: true },
|
||||||
|
}));
|
||||||
|
deletedCount += chunk.length;
|
||||||
|
this.ctx.logger.info(`[UnifiedStorageProvider] ${chunk.length} objetos deletados do Wasabi (lote seletivo)`);
|
||||||
|
} catch (err: any) {
|
||||||
|
this.ctx.logger.error(`[UnifiedStorageProvider] Erro no delete em lote do Wasabi: ${err.message}`);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// 2. Fallback local
|
||||||
|
for (const key of keys) {
|
||||||
|
const localPath = path.join(this.localFallbackDir, key);
|
||||||
|
try {
|
||||||
|
if (fs.existsSync(localPath)) fs.unlinkSync(localPath);
|
||||||
|
} catch (err: any) {
|
||||||
|
this.ctx.logger.error(`[UnifiedStorageProvider] Erro ao remover fallback local ${key}: ${err.message}`);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return { deleted: deletedCount };
|
||||||
|
}
|
||||||
|
|
||||||
async upload(payload: StorageUploadPayload): Promise<{ provider: string; path: string }> {
|
async upload(payload: StorageUploadPayload): Promise<{ provider: string; path: string }> {
|
||||||
const { category, file } = payload;
|
const { category, file } = payload;
|
||||||
if (!VALID_CATEGORIES.includes(category)) throw new Error(`Categoria inválida: ${category}`);
|
if (!VALID_CATEGORIES.includes(category)) throw new Error(`Categoria inválida: ${category}`);
|
||||||
|
|||||||
Reference in New Issue
Block a user