feat(attachments): cleaning and setting up module for app integration
This commit is contained in:
Matthieu Haineault
parent
fb0187c117
commit
4b0189ed48
|
|
@ -1,22 +1,26 @@
|
|||
import { ArchivalAttachmentService } from "src/modules/attachments/services/archival-attachment.service";
|
||||
import { AttachmentArchivalService } from "src/modules/attachments/services/attachment-archival.service";
|
||||
import { GarbargeCollectorService } from "src/modules/attachments/services/garbage-collector.service";
|
||||
import { AttachmentsController } from "src/modules/attachments/controllers/attachments.controller";
|
||||
import { DiskStorageService } from "src/modules/attachments/services/disk-storage.service";
|
||||
// import { ScheduleModule } from "@nestjs/schedule";
|
||||
import { VariantsQueue } from "src/modules/attachments/services/variants.queue";
|
||||
import { Module } from "@nestjs/common";
|
||||
import { AttachmentDeleteService } from "src/modules/attachments/services/attachment-delete.service";
|
||||
import { AttachmentUploadService } from "src/modules/attachments/services/attachment-upload.service";
|
||||
import { AttachmentGetService } from "src/modules/attachments/services/attachment-get.service";
|
||||
|
||||
@Module({
|
||||
// imports: [ScheduleModule.forRoot()],
|
||||
controllers: [ AttachmentsController],
|
||||
providers: [
|
||||
ArchivalAttachmentService,
|
||||
AttachmentArchivalService,
|
||||
GarbargeCollectorService,
|
||||
DiskStorageService,
|
||||
VariantsQueue,
|
||||
AttachmentDeleteService,
|
||||
AttachmentUploadService,
|
||||
AttachmentGetService,
|
||||
],
|
||||
exports: [
|
||||
ArchivalAttachmentService,
|
||||
AttachmentArchivalService,
|
||||
GarbargeCollectorService
|
||||
],
|
||||
})
|
||||
|
|
|
|||
|
|
@ -2,4 +2,5 @@ export const maxUploadBytes = () =>
|
|||
(Number(process.env.MAX_UPLOAD_MB || 25)) * 1024 * 1024;
|
||||
|
||||
export const allowedMimes = () =>
|
||||
(process.env.ALLOWED_MIME || 'image/jpeg,image/png,image/webp,application/pdf').split(',').map(s =>s.trim()).filter(Boolean);
|
||||
(process.env.ALLOWED_MIME || 'image/jpeg,image/png,image/webp,application/pdf')
|
||||
.split(',').map(s => s.trim()).filter(Boolean);
|
||||
|
|
@ -1,212 +1,45 @@
|
|||
import { Controller, UseInterceptors, Post, Get, Param, Res, UploadedFile, Body, Delete, Query } from "@nestjs/common";
|
||||
import { FileInterceptor } from "@nestjs/platform-express";
|
||||
import { DiskStorageService } from "../services/disk-storage.service";
|
||||
import {
|
||||
Controller,NotFoundException, UseInterceptors, Post, Get, Param, Res,
|
||||
UploadedFile, BadRequestException, UnsupportedMediaTypeException, Body, Delete,
|
||||
Query,
|
||||
} from "@nestjs/common";
|
||||
import { maxUploadBytes, allowedMimes } from "../config/upload.config";
|
||||
import { memoryStorage } from 'multer';
|
||||
import { fileTypeFromBuffer, fileTypeFromFile } from "file-type";
|
||||
import { Readable } from "node:stream";
|
||||
import { PrismaService } from "src/prisma/prisma.service";
|
||||
import { AttachmentDeleteService } from "src/modules/attachments/services/attachment-delete.service";
|
||||
import { AttachmentUploadService } from "src/modules/attachments/services/attachment-upload.service";
|
||||
import { AttachmentGetService } from "src/modules/attachments/services/attachment-get.service";
|
||||
import { UploadMetaAttachmentsDto } from "../dtos/upload-meta-attachments.dto";
|
||||
import { resolveAttachmentsRoot } from "src/config/attachment.config";
|
||||
import * as path from 'node:path';
|
||||
import { promises as fsp } from 'node:fs';
|
||||
import { createReadStream } from "node:fs";
|
||||
import { AdminSearchDto } from "../dtos/search-filters.dto";
|
||||
import { maxUploadBytes } from "../config/upload.config";
|
||||
import { memoryStorage } from 'multer';
|
||||
import { Response } from 'express';
|
||||
import { VariantsQueue } from "../services/variants.queue";
|
||||
import { AdminSearchDto } from "../dtos/admin-search.dto";
|
||||
|
||||
@Controller('attachments')
|
||||
export class AttachmentsController {
|
||||
constructor(
|
||||
private readonly disk: DiskStorageService,
|
||||
private readonly prisma: PrismaService,
|
||||
private readonly variantsQ: VariantsQueue,
|
||||
) {}
|
||||
private readonly uploadService: AttachmentUploadService,
|
||||
private readonly deleteService: AttachmentDeleteService,
|
||||
private readonly getService: AttachmentGetService,
|
||||
) { }
|
||||
|
||||
@Get(':id')
|
||||
async getById(
|
||||
@Param('id') id: string,
|
||||
@Query('variant') variant: string | undefined,
|
||||
@Res() res: Response,
|
||||
) {
|
||||
const num_id = Number(id);
|
||||
if(!Number.isFinite(num_id)) throw new NotFoundException('Invalid id');
|
||||
|
||||
const attachment = await this.prisma.attachments.findUnique({
|
||||
where: { id: num_id },
|
||||
include: { blob: true },
|
||||
});
|
||||
if (!attachment) throw new NotFoundException();
|
||||
|
||||
const relative = variant ? `${attachment.blob.storage_path}.${variant}` : attachment.blob.storage_path;
|
||||
const abs = path.join(resolveAttachmentsRoot(), relative);
|
||||
|
||||
let stat;
|
||||
try {
|
||||
stat = await fsp.stat(abs);
|
||||
}catch {
|
||||
throw new NotFoundException('File not found');
|
||||
}
|
||||
|
||||
let mime = attachment.blob.mime;
|
||||
try {
|
||||
const kind = await fileTypeFromFile(abs);
|
||||
if(kind?.mime) mime = kind.mime;
|
||||
} catch {}
|
||||
res.set('Content-Type', mime);
|
||||
res.set('Content-Length', String(stat.size));
|
||||
res.set('ETag', `"sha256-${attachment.blob.sha256}${variant ? '.'+variant : ''}"`);
|
||||
res.set('Last-Modified', stat.mtime.toUTCString());
|
||||
res.set('Cache-Control', 'private, max-age=31536000, immutable');
|
||||
res.set('X-Content-Type-Options', 'nosniff');
|
||||
|
||||
createReadStream(abs).pipe(res);
|
||||
async getById(@Param('id') id: string, @Query('variant') variant: string | undefined, @Res() res: Response) {
|
||||
return await this.getService.findAttachmentById(id, variant, res)
|
||||
}
|
||||
|
||||
@Get('variants/:id')
|
||||
async listVariants(@Param('id')id: string) {
|
||||
const num_id = Number(id);
|
||||
if(!Number.isFinite(num_id)) throw new NotFoundException('Invalid variant id');
|
||||
return this.prisma.attachmentVariants.findMany({
|
||||
where: { attachment_id: num_id },
|
||||
orderBy: { variant: 'asc'},
|
||||
select: { variant: true, bytes: true, width: true, height: true, path: true, created_at: true },
|
||||
});
|
||||
async getlistVariantsById(@Param('id') id: string) {
|
||||
return await this.getService.getListVariants(id);
|
||||
}
|
||||
|
||||
@Delete(':id')
|
||||
async remove(@Param('id') id: string) {
|
||||
const result = await this.prisma.$transaction(async (tx) => {
|
||||
const att = await tx.attachments.findUnique({ where: { id: Number(id) } });
|
||||
if (!att) throw new NotFoundException();
|
||||
|
||||
// soft-delete
|
||||
await tx.attachments.update({
|
||||
where: { id: Number(id) },
|
||||
data: { status: 'DELETED' },
|
||||
});
|
||||
|
||||
// decrement refcount
|
||||
const dec = await tx.$executeRaw
|
||||
`UPDATE "Blobs" SET refcount = refcount - 1
|
||||
WHERE sha256 = ${att.sha256} AND refcount > 0`
|
||||
;
|
||||
|
||||
return { ok: true, decremented: dec > 0 };
|
||||
});
|
||||
return result;
|
||||
return await this.deleteService.deleteAttachment(id);
|
||||
}
|
||||
|
||||
|
||||
@Post()
|
||||
@UseInterceptors(
|
||||
FileInterceptor('file', { storage: memoryStorage(), limits: { fileSize: maxUploadBytes() }})
|
||||
)
|
||||
async upload(
|
||||
@UploadedFile() file?: Express.Multer.File,
|
||||
@Body() meta?: UploadMetaAttachmentsDto,
|
||||
) {
|
||||
if(!file) throw new BadRequestException('No file found');
|
||||
|
||||
//magic detection using binary signature
|
||||
const kind = await fileTypeFromBuffer(file.buffer).catch(() => null);
|
||||
const detected_mime = kind?.mime || file.mimetype || 'application/octet-stream';
|
||||
|
||||
//strict whitelist
|
||||
if(!allowedMimes().includes(detected_mime)) {
|
||||
throw new UnsupportedMediaTypeException(`This type is not supported: ${detected_mime}`);
|
||||
}
|
||||
|
||||
//Saving FS (hash + CAS + unDupes)
|
||||
const stream = Readable.from(file.buffer);
|
||||
const { sha256, storage_path, size } = await this.disk.saveStreamAndHash(stream);
|
||||
|
||||
const now = new Date();
|
||||
const attachment = await this.prisma.$transaction(async (tx) => {
|
||||
//upsert blob: +1 ref
|
||||
await tx.blobs.upsert({
|
||||
where: { sha256 },
|
||||
create: {
|
||||
sha256,
|
||||
storage_path: storage_path,
|
||||
size,
|
||||
mime: detected_mime,
|
||||
refcount: 1,
|
||||
created_at: now,
|
||||
},
|
||||
update: { //only increment, does not change the storage path
|
||||
refcount: { increment: 1 },
|
||||
mime: detected_mime, //update mime and size to keep last image
|
||||
size,
|
||||
},
|
||||
});
|
||||
|
||||
const att = await tx.attachments.create({
|
||||
data: {
|
||||
sha256,
|
||||
owner_type: meta?.owner_type ?? 'EXPENSE',
|
||||
owner_id: meta?.owner_id ?? 'unknown',
|
||||
original_name: file.originalname,
|
||||
status: 'ACTIVE',
|
||||
retention_policy: (meta?.retention_policy ?? 'EXPENSE_7Y') as any,
|
||||
created_by: meta?.created_by ?? 'system',
|
||||
created_at: now,
|
||||
},
|
||||
});
|
||||
return att;
|
||||
});
|
||||
|
||||
await this.variantsQ.enqueue(attachment.id, detected_mime);
|
||||
|
||||
return {
|
||||
ok: true,
|
||||
id: attachment.id,
|
||||
sha256,
|
||||
storage_path: storage_path,
|
||||
size,
|
||||
mime: detected_mime,
|
||||
original_name: file.originalname,
|
||||
owner_type: attachment.owner_type,
|
||||
owner_id: attachment.owner_id,
|
||||
};
|
||||
@UseInterceptors(FileInterceptor('file', { storage: memoryStorage(), limits: { fileSize: maxUploadBytes() } }))
|
||||
async upload(@UploadedFile() file?: Express.Multer.File, @Body() meta?: UploadMetaAttachmentsDto) {
|
||||
return await this.uploadService.uploadAttachment(file, meta);
|
||||
}
|
||||
|
||||
@Get('/admin/search')
|
||||
async adminSearch(
|
||||
@Query() query: AdminSearchDto ) {
|
||||
const where: any = {};
|
||||
if (query.owner_type) where.owner_type = query.owner_type;
|
||||
if (query.owner_id) where.owner_id = query.owner_id;
|
||||
|
||||
if (query.date_from || query.date_to) {
|
||||
where.created_at = {};
|
||||
if (query.date_from) where.created_at.gte = new Date(query.date_from + 'T00:00:00Z');
|
||||
if (query.date_to) where.created_at.lte = new Date(query.date_to + 'T23:59:59Z');
|
||||
}
|
||||
|
||||
const page = query.page ?? 1;
|
||||
const page_size = query.page_size ?? 50;
|
||||
const skip = (page - 1)* page_size;
|
||||
const take = page_size;
|
||||
|
||||
const [items, total] = await this.prisma.$transaction([
|
||||
this.prisma.attachments.findMany({
|
||||
where,
|
||||
orderBy: { created_at: 'desc' },
|
||||
skip, take,
|
||||
include: {
|
||||
blob: {
|
||||
select: { mime: true, size: true, storage_path: true, sha256: true },
|
||||
},
|
||||
},
|
||||
}),
|
||||
this.prisma.attachments.count({ where }),
|
||||
]);
|
||||
|
||||
return { page, page_size: take, total, items };
|
||||
@Get('search/filters')
|
||||
async searchWithFilters(@Query() dto: AdminSearchDto) {
|
||||
return await this.getService.searchAttachmentWithFilters(dto);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,34 +0,0 @@
|
|||
import { Type } from "class-transformer";
|
||||
import { IsInt, IsISO8601, IsOptional, IsString, Max, Min } from "class-validator";
|
||||
|
||||
export class AdminSearchDto {
|
||||
|
||||
@IsOptional()
|
||||
@IsString()
|
||||
owner_type?: string;
|
||||
|
||||
@IsOptional()
|
||||
@IsString()
|
||||
owner_id?: string;
|
||||
|
||||
@IsOptional()
|
||||
@IsISO8601()
|
||||
date_from?: string;
|
||||
|
||||
@IsOptional()
|
||||
@IsISO8601()
|
||||
date_to?: string;
|
||||
|
||||
@IsOptional()
|
||||
@Type(()=> Number)
|
||||
@IsInt()
|
||||
@Min(1)
|
||||
page?: number = 1;
|
||||
|
||||
@IsOptional()
|
||||
@Type(()=> Number)
|
||||
@IsInt()
|
||||
@Min(1)
|
||||
@Max(200)
|
||||
page_size?: number = 50;
|
||||
}
|
||||
10
src/modules/attachments/dtos/search-filters.dto.ts
Normal file
10
src/modules/attachments/dtos/search-filters.dto.ts
Normal file
|
|
@ -0,0 +1,10 @@
|
|||
import { IsInt, IsOptional, IsString, Max, Min } from "class-validator";
|
||||
|
||||
export class AdminSearchDto {
|
||||
@IsOptional() @IsString() owner_type?: string;
|
||||
@IsOptional() @IsString() owner_id?: string;
|
||||
@IsOptional() date_from?: string;
|
||||
@IsOptional() date_to?: string;
|
||||
@IsOptional() @IsInt() @Min(1) page?: number = 1;
|
||||
@IsOptional() @IsInt() @Min(1) @Max(200) page_size?: number = 50;
|
||||
}
|
||||
|
|
@ -1,45 +1,41 @@
|
|||
import { Injectable, Logger } from "@nestjs/common";
|
||||
import { Injectable } from "@nestjs/common";
|
||||
import { Cron } from "@nestjs/schedule";
|
||||
import { startOfYear } from "src/modules/attachments/utils/cas.util";
|
||||
import { PrismaService } from "src/prisma/prisma.service";
|
||||
|
||||
@Injectable()
|
||||
export class ArchivalAttachmentService {
|
||||
private readonly logger = new Logger(ArchivalAttachmentService.name)
|
||||
export class AttachmentArchivalService {
|
||||
private readonly batch_size = Number(process.env.ARCHIVE_BATCH_SIZE || 1000);
|
||||
private readonly cron_expression = process.env.ARCHIVE_CRON || '0 3 * * 1';
|
||||
|
||||
constructor( private readonly prisma: PrismaService) {}
|
||||
constructor(private readonly prisma: PrismaService) { }
|
||||
|
||||
private startOfYear(): Date {
|
||||
const now = new Date();
|
||||
return new Date(Date.UTC(now.getUTCFullYear(), 0, 1, 0, 0, 0, 0));
|
||||
}
|
||||
|
||||
@Cron(function (this: ArchivalAttachmentService) { return this.cron_expression; } as any)
|
||||
@Cron(function (this: AttachmentArchivalService) { return this.cron_expression; } as any)
|
||||
async runScheduled() {
|
||||
await this.archiveCutoffToStartOfYear();
|
||||
}
|
||||
|
||||
//archive everything before current year
|
||||
async archiveCutoffToStartOfYear() {
|
||||
const cutoff = this.startOfYear();
|
||||
this.logger.log(`Archival: cutoff=${cutoff.toISOString()} batch=${this.batch_size}`);
|
||||
const cutoff = startOfYear();
|
||||
console.log(`Archival: cutoff=${cutoff.toISOString()} batch=${this.batch_size}`);
|
||||
|
||||
let moved = 0, total = 0, i = 0;
|
||||
do {
|
||||
moved = await this.archiveBatch(cutoff, this.batch_size);
|
||||
total += moved;
|
||||
i++;
|
||||
if(moved > 0) this.logger.log(`Batch #${i}: moved ${moved}`);
|
||||
}while (moved === this.batch_size);
|
||||
if (moved > 0) console.log(`Batch #${i}: moved ${moved}`);
|
||||
} while (moved === this.batch_size);
|
||||
|
||||
this.logger.log(`Archival done: total moved : ${total}`);
|
||||
console.log(`Archival done: total moved : ${total}`);
|
||||
return { moved: total };
|
||||
}
|
||||
|
||||
//only moves table content to archive and not blobs.
|
||||
private async archiveBatch(cutoff: Date, batch_size: number): Promise<number> {
|
||||
const moved = await this.prisma.$executeRaw`
|
||||
const moved = await this.prisma.$executeRaw `
|
||||
WITH moved AS (
|
||||
DELETE FROM "attachments"
|
||||
WHERE id IN (
|
||||
|
|
@ -52,9 +48,8 @@ export class ArchivalAttachmentService {
|
|||
)
|
||||
INSERT INTO archive.attachments_archive
|
||||
(id, sha256, owner_type, owner_id, original_name, status, retention_policy, created_by, created_at)
|
||||
SELECT * FROM moved;
|
||||
`;
|
||||
return Number(moved) || 0;
|
||||
SELECT * FROM moved;`;
|
||||
return Number(moved) || 0;
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,26 @@
|
|||
import { Injectable, NotFoundException } from "@nestjs/common";
|
||||
import { Result } from "src/common/errors/result-error.factory";
|
||||
import { PrismaService } from "src/prisma/prisma.service";
|
||||
|
||||
@Injectable()
|
||||
export class AttachmentDeleteService {
|
||||
constructor(private readonly prisma: PrismaService) { }
|
||||
|
||||
async deleteAttachment(id: string): Promise<Result<boolean, string>> {
|
||||
await this.prisma.$transaction(async (tx) => {
|
||||
const attachment = await tx.attachments.findUnique({ where: { id: Number(id) } });
|
||||
if (!attachment) return { success: false, error: 'ATTACHMENT_NOT_FOUND' };
|
||||
|
||||
// soft-delete
|
||||
await tx.attachments.update({ where: { id: Number(id) }, data: { status: 'DELETED' } });
|
||||
|
||||
// decrement refcount
|
||||
const dec = await tx.$executeRaw `
|
||||
UPDATE "Blobs" SET refcount = refcount - 1
|
||||
WHERE sha256 = ${attachment.sha256} AND refcount > 0;`;
|
||||
|
||||
return { ok: true, decremented: dec > 0 };
|
||||
});
|
||||
return { success: true, data: true };
|
||||
}
|
||||
}
|
||||
100
src/modules/attachments/services/attachment-get.service.ts
Normal file
100
src/modules/attachments/services/attachment-get.service.ts
Normal file
|
|
@ -0,0 +1,100 @@
|
|||
import { Injectable } from "@nestjs/common";
|
||||
import { Response } from "express";
|
||||
import { AdminSearchDto } from "src/modules/attachments/dtos/search-filters.dto";
|
||||
import { PrismaService } from "src/prisma/prisma.service";
|
||||
import { resolveAttachmentsRoot } from "src/config/attachment.config";
|
||||
import * as path from 'node:path';
|
||||
import { promises as fsp } from 'node:fs';
|
||||
import { createReadStream } from "node:fs";
|
||||
import { fileTypeFromFile } from "file-type";
|
||||
import { Result } from "src/common/errors/result-error.factory";
|
||||
|
||||
@Injectable()
|
||||
export class AttachmentGetService {
|
||||
constructor(
|
||||
private readonly prisma: PrismaService,
|
||||
|
||||
) { }
|
||||
|
||||
async getListVariants(id: string): Promise<Result<any, string>> {
|
||||
const num_id = Number(id);
|
||||
if (!Number.isFinite(num_id)) return { success: false, error: 'INVALID_ATTACHMENTS' };
|
||||
const variants = await this.prisma.attachmentVariants.findMany({
|
||||
where: { attachment_id: num_id },
|
||||
orderBy: { variant: 'asc' },
|
||||
select: { variant: true, bytes: true, width: true, height: true, path: true, created_at: true },
|
||||
});
|
||||
return { success: true, data: variants };
|
||||
}
|
||||
|
||||
async searchAttachmentWithFilters(dto: AdminSearchDto): Promise<Result<any, string>> {
|
||||
const where: any = {};
|
||||
if (dto.owner_type) where.owner_type = dto.owner_type;
|
||||
if (dto.owner_id) where.owner_id = dto.owner_id;
|
||||
|
||||
if (dto.date_from || dto.date_to) {
|
||||
where.created_at = {};
|
||||
if (dto.date_from) where.created_at.gte = new Date(dto.date_from + 'T00:00:00Z');
|
||||
if (dto.date_to) where.created_at.lte = new Date(dto.date_to + 'T23:59:59Z');
|
||||
}
|
||||
|
||||
const page = dto.page ?? 1;
|
||||
const page_size = dto.page_size ?? 50;
|
||||
const skip = (page - 1) * page_size;
|
||||
const take = page_size;
|
||||
|
||||
const [items, total] = await this.prisma.$transaction([
|
||||
this.prisma.attachments.findMany({
|
||||
where,
|
||||
orderBy: { created_at: 'desc' },
|
||||
skip, take,
|
||||
include: {
|
||||
blob: {
|
||||
select: { mime: true, size: true, storage_path: true, sha256: true },
|
||||
},
|
||||
},
|
||||
}),
|
||||
this.prisma.attachments.count({ where }),
|
||||
]);
|
||||
|
||||
return { success: true, data: { page, page_size: take, total, items } };
|
||||
}
|
||||
|
||||
|
||||
async findAttachmentById(id: string, variant: string | undefined, res: Response): Promise<Result<boolean, string>> {
|
||||
const num_id = Number(id);
|
||||
if (!Number.isFinite(num_id)) return { success: false, error: 'INVALID_ATTACHMENTS' };
|
||||
|
||||
const attachment = await this.prisma.attachments.findUnique({
|
||||
where: { id: num_id },
|
||||
include: { blob: true },
|
||||
});
|
||||
if (!attachment) return { success: false, error: 'ATTACHMENT_NOT_FOUND' };
|
||||
|
||||
const relative = variant ? `${attachment.blob.storage_path}.${variant}` : attachment.blob.storage_path;
|
||||
const abs = path.join(resolveAttachmentsRoot(), relative);
|
||||
|
||||
let stat;
|
||||
try {
|
||||
stat = await fsp.stat(abs);
|
||||
} catch {
|
||||
return { success: false, error: 'INVALID_FILE_PATH' };
|
||||
}
|
||||
|
||||
let mime = attachment.blob.mime;
|
||||
try {
|
||||
const kind = await fileTypeFromFile(abs);
|
||||
if (kind?.mime) mime = kind.mime;
|
||||
} catch { }
|
||||
res.set('Content-Type', mime);
|
||||
res.set('Content-Length', String(stat.size));
|
||||
res.set('ETag', `"sha256-${attachment.blob.sha256}${variant ? '.' + variant : ''}"`);
|
||||
res.set('Last-Modified', stat.mtime.toUTCString());
|
||||
res.set('Cache-Control', 'private, max-age=31536000, immutable');
|
||||
res.set('X-Content-Type-Options', 'nosniff');
|
||||
|
||||
createReadStream(abs).pipe(res);
|
||||
return { success: true, data: true };
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,87 @@
|
|||
import { Injectable } from "@nestjs/common";
|
||||
import { allowedMimes } from "src/modules/attachments/config/upload.config";
|
||||
import { UploadMetaAttachmentsDto } from "src/modules/attachments/dtos/upload-meta-attachments.dto";
|
||||
import { Readable } from "node:stream";
|
||||
import { PrismaService } from "src/prisma/prisma.service";
|
||||
import { fileTypeFromBuffer } from "file-type";
|
||||
import { DiskStorageService } from "src/modules/attachments/services/disk-storage.service";
|
||||
import { VariantsQueue } from "src/modules/attachments/services/variants.queue";
|
||||
import { Result } from "src/common/errors/result-error.factory";
|
||||
|
||||
@Injectable()
|
||||
export class AttachmentUploadService {
|
||||
constructor(
|
||||
private readonly prisma: PrismaService,
|
||||
private readonly disk: DiskStorageService,
|
||||
private readonly variantsQ: VariantsQueue,
|
||||
) { }
|
||||
|
||||
async uploadAttachment(file?: Express.Multer.File, meta?: UploadMetaAttachmentsDto): Promise<Result<any, string>> {
|
||||
if (!file) return { success: false, error: 'FILE_NOT_FOUND' };
|
||||
|
||||
//magic detection using binary signature
|
||||
const kind = await fileTypeFromBuffer(file.buffer).catch(() => null);
|
||||
const detected_mime = kind?.mime || file.mimetype || 'application/octet-stream';
|
||||
|
||||
//strict whitelist
|
||||
if (!allowedMimes().includes(detected_mime)) {
|
||||
return { success: false, error: 'INVALID_ATTACHMENT_TYPE' };
|
||||
}
|
||||
|
||||
//Saving FS (hash + CAS + unDupes)
|
||||
const stream = Readable.from(file.buffer);
|
||||
const { sha256, storage_path, size } = await this.disk.saveStreamAndHash(stream);
|
||||
|
||||
const now = new Date();
|
||||
const attachment = await this.prisma.$transaction(async (tx) => {
|
||||
//upsert blob: +1 ref
|
||||
await tx.blobs.upsert({
|
||||
where: { sha256 },
|
||||
create: {
|
||||
sha256,
|
||||
storage_path: storage_path,
|
||||
size,
|
||||
mime: detected_mime,
|
||||
refcount: 1,
|
||||
created_at: now,
|
||||
},
|
||||
update: { //only increment, does not change the storage path
|
||||
refcount: { increment: 1 },
|
||||
mime: detected_mime, //update mime and size to keep last image
|
||||
size,
|
||||
},
|
||||
});
|
||||
|
||||
const att = await tx.attachments.create({
|
||||
data: {
|
||||
sha256,
|
||||
owner_type: meta?.owner_type ?? 'EXPENSE',
|
||||
owner_id: meta?.owner_id ?? 'unknown',
|
||||
original_name: file.originalname,
|
||||
status: 'ACTIVE',
|
||||
retention_policy: (meta?.retention_policy ?? 'EXPENSE_7Y') as any,
|
||||
created_by: meta?.created_by ?? 'system',
|
||||
created_at: now,
|
||||
},
|
||||
});
|
||||
return att;
|
||||
});
|
||||
|
||||
await this.variantsQ.enqueue(attachment.id, detected_mime);
|
||||
|
||||
return {
|
||||
success: true,
|
||||
data: {
|
||||
ok: true,
|
||||
id: attachment.id,
|
||||
sha256,
|
||||
storage_path: storage_path,
|
||||
size,
|
||||
mime: detected_mime,
|
||||
original_name: file.originalname,
|
||||
owner_type: attachment.owner_type,
|
||||
owner_id: attachment.owner_id,
|
||||
}
|
||||
};
|
||||
}
|
||||
}
|
||||
|
|
@ -4,32 +4,21 @@ import { promises as fsp } from 'node:fs';
|
|||
import { createWriteStream, statSync, existsSync } from 'node:fs';
|
||||
import { join, dirname } from 'node:path';
|
||||
import { pipeline } from 'node:stream/promises';
|
||||
import { ATT_TMP_DIR, resolveAttachmentsRoot } from 'src/config/attachment.config';
|
||||
import { ATT_TMP_DIR } from 'src/config/attachment.config';
|
||||
import { casPathFor, getAbsolutePath } from 'src/modules/attachments/utils/cas.util';
|
||||
|
||||
export type SaveResult = { sha256:string, storage_path:string, size:number};
|
||||
export type SaveResult = { sha256: string, storage_path: string, size: number };
|
||||
|
||||
@Injectable()
|
||||
export class DiskStorageService {
|
||||
private root = resolveAttachmentsRoot();
|
||||
|
||||
private casPath(hash: string) {
|
||||
const a = hash.slice(0,2), b = hash.slice(2,4);
|
||||
return `sha256/${a}/${b}/${hash}`;
|
||||
}
|
||||
|
||||
//chemin absolue du storage
|
||||
getAbsolutePath(storagePathRel: string) {
|
||||
return join(this.root, storagePathRel);
|
||||
}
|
||||
|
||||
async exists(storagePathRel: string) {
|
||||
try {
|
||||
statSync(this.getAbsolutePath(storagePathRel));
|
||||
return true;
|
||||
}catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
// async exists(storagePathRel: string) {
|
||||
// try {
|
||||
// statSync(getAbsolutePath(storagePathRel));
|
||||
// return true;
|
||||
// } catch {
|
||||
// return false;
|
||||
// }
|
||||
// }
|
||||
|
||||
//adds file and hash it
|
||||
async saveStreamAndHash(input: NodeJS.ReadableStream): Promise<SaveResult> {
|
||||
|
|
@ -44,25 +33,25 @@ export class DiskStorageService {
|
|||
await pipeline(input, tmpOut); //await end of writing stream
|
||||
|
||||
const sha = hash.digest('hex');
|
||||
const rel = this.casPath(sha);
|
||||
const finalAbs = this.getAbsolutePath(rel);
|
||||
const rel = casPathFor(sha);
|
||||
const finalAbs = getAbsolutePath(rel);
|
||||
|
||||
// 2- is there is no destination => move (atomic renaming on the same volume)
|
||||
if(!existsSync(finalAbs)) {
|
||||
await fsp.mkdir(dirname(finalAbs), { recursive:true });
|
||||
if (!existsSync(finalAbs)) {
|
||||
await fsp.mkdir(dirname(finalAbs), { recursive: true });
|
||||
try {
|
||||
await fsp.rename(tmpPath, finalAbs);
|
||||
}catch (e) {
|
||||
} catch (e) {
|
||||
//if someone is faster and used the same hash
|
||||
if(existsSync(finalAbs)) {
|
||||
await fsp.rm(tmpPath, { force:true });
|
||||
if (existsSync(finalAbs)) {
|
||||
await fsp.rm(tmpPath, { force: true });
|
||||
} else {
|
||||
throw e;
|
||||
}
|
||||
}
|
||||
} else {
|
||||
//remove duplicata if already exists
|
||||
await fsp.rm(tmpPath, { force:true });
|
||||
await fsp.rm(tmpPath, { force: true });
|
||||
}
|
||||
|
||||
const size = statSync(finalAbs).size;
|
||||
|
|
|
|||
|
|
@ -16,10 +16,10 @@ export class GarbargeCollectorService {
|
|||
//fetchs root of storage
|
||||
private readonly root = resolveAttachmentsRoot();
|
||||
|
||||
constructor(private readonly prisma: PrismaService) {}
|
||||
constructor(private readonly prisma: PrismaService) { }
|
||||
|
||||
//planif for the Cronjob
|
||||
@Cron(function(this:GarbargeCollectorService) { return this.cron_expression; } as any)
|
||||
@Cron(function (this: GarbargeCollectorService) { return this.cron_expression; } as any)
|
||||
async runScheduled() {
|
||||
await this.collect();
|
||||
}
|
||||
|
|
@ -28,15 +28,15 @@ export class GarbargeCollectorService {
|
|||
async collect() {
|
||||
let total = 0, round = 0;
|
||||
//infinit loop (;;) with break
|
||||
for(;;) {
|
||||
for (; ;) {
|
||||
round++;
|
||||
const num = await this.collectBatch();
|
||||
total += num;
|
||||
this.logger.log(`Garbage Collector round #${round} removed ${num}`);
|
||||
if(num < this.batch_size) break; //breaks if not a full batch
|
||||
if (num < this.batch_size) break; //breaks if not a full batch
|
||||
}
|
||||
this.logger.log(`Garbage Collecting done: total removed ${total}`);
|
||||
return { removed:total };
|
||||
return { removed: total };
|
||||
}
|
||||
|
||||
//Manage a single lot of orphan blobs
|
||||
|
|
@ -46,33 +46,33 @@ export class GarbargeCollectorService {
|
|||
select: { sha256: true, storage_path: true },
|
||||
take: this.batch_size,
|
||||
});
|
||||
if(blobs.length === 0) return 0;
|
||||
if (blobs.length === 0) return 0;
|
||||
|
||||
// delete original file and all its variants <hash> in the same file
|
||||
await Promise.all(
|
||||
blobs.map(async (blob)=> {
|
||||
blobs.map(async (blob) => {
|
||||
const absolute_path = path.join(this.root, blob.storage_path);
|
||||
await this.deleteFileIfExists(absolute_path); //tries to delete original file if found
|
||||
|
||||
const dir = path.dirname(absolute_path);
|
||||
const base = path.basename(absolute_path);
|
||||
try {
|
||||
const entries = await fsp.readdir(dir, { withFileTypes: true});
|
||||
const entries = await fsp.readdir(dir, { withFileTypes: true });
|
||||
const targets = entries.filter(entry => entry.isFile() && entry.name.startsWith(base + '.'))
|
||||
.map(entry => path.join(dir, entry.name));
|
||||
.map(entry => path.join(dir, entry.name));
|
||||
//deletes all variants
|
||||
await Promise.all(targets.map(target => this.deleteFileIfExists(target)));
|
||||
} catch {}
|
||||
} catch { }
|
||||
})
|
||||
);
|
||||
//deletes blobs lignes if file is deleted
|
||||
const hashes = blobs.map(blob => blob.sha256);
|
||||
await this.prisma.blobs.deleteMany({where: { sha256: { in: hashes } } });
|
||||
await this.prisma.blobs.deleteMany({ where: { sha256: { in: hashes } } });
|
||||
return blobs.length;
|
||||
}
|
||||
|
||||
//helper: deletes path if exists and ignore errors
|
||||
private async deleteFileIfExists(path: string) {
|
||||
try { await fsp.unlink(path); } catch {}
|
||||
try { await fsp.unlink(path); } catch { }
|
||||
}
|
||||
}
|
||||
|
|
@ -3,7 +3,7 @@ import { Queue } from "bullmq";
|
|||
|
||||
@Injectable()
|
||||
export class VariantsQueue {
|
||||
private queue : Queue;
|
||||
private queue: Queue;
|
||||
|
||||
constructor() {
|
||||
const name = `${process.env.BULL_PREFIX || 'attachments'}:variants`;
|
||||
|
|
@ -11,12 +11,11 @@ export class VariantsQueue {
|
|||
}
|
||||
|
||||
enqueue(attachment_id: number, mime: string) {
|
||||
if(!mime.startsWith('image/')) {
|
||||
return Promise.resolve();
|
||||
}
|
||||
return this.queue.add('generate',
|
||||
{ attachment_id, mime },
|
||||
{ attempts: 3, backoff: { type: 'exponential', delay:2000 } }
|
||||
if (!mime.startsWith('image/')) return Promise.resolve();
|
||||
return this.queue.add(
|
||||
'generate',
|
||||
{ attachment_id, mime },
|
||||
{ attempts: 3, backoff: { type: 'exponential', delay: 2000 } }
|
||||
);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,4 +1,17 @@
|
|||
import { join } from "node:path";
|
||||
|
||||
export function casPathFor(hash: string) {
|
||||
const a = hash.slice(0,2), b = hash.slice(2,4);
|
||||
const a = hash.slice(0, 2), b = hash.slice(2, 4);
|
||||
return `sha256/${a}/${b}/${hash}`;
|
||||
}
|
||||
|
||||
//chemin absolue du storage
|
||||
export function getAbsolutePath(storagePathRel: string) {
|
||||
return join(this.root, storagePathRel);
|
||||
}
|
||||
|
||||
|
||||
export function startOfYear(): Date {
|
||||
const now = new Date();
|
||||
return new Date(Date.UTC(now.getUTCFullYear(), 0, 1, 0, 0, 0, 0));
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user