From caef31c7039d55c431340344ce2efdf44407a475 Mon Sep 17 00:00:00 2001 From: asepharyana Date: Mon, 6 Jul 2026 14:23:37 +0700 Subject: [PATCH] feat: add DB CRUD layer for buckets, multipart, and S3 file extensions --- src/db/buckets.ts | 67 +++++++++++++++++++++++++ src/db/files-ext.ts | 118 ++++++++++++++++++++++++++++++++++++++++++++ src/db/multipart.ts | 97 ++++++++++++++++++++++++++++++++++++ src/db/schema.ts | 7 ++- 4 files changed, 288 insertions(+), 1 deletion(-) create mode 100644 src/db/buckets.ts create mode 100644 src/db/files-ext.ts create mode 100644 src/db/multipart.ts diff --git a/src/db/buckets.ts b/src/db/buckets.ts new file mode 100644 index 0000000..1b2ad55 --- /dev/null +++ b/src/db/buckets.ts @@ -0,0 +1,67 @@ +import { sql } from 'drizzle-orm'; +import { db } from './index'; + +export interface Bucket { + id: string; + name: string; + createdAt: Date; + updatedAt: Date; +} + +interface QueryResult { + rows: Record[]; + rowCount: number; +} + +export const createBucket = async (name: string): Promise => { + const result = (await db.execute( + sql`INSERT INTO buckets (name) VALUES (${name}) RETURNING id, name, created_at, updated_at`, + )) as unknown as QueryResult; + const row = result.rows[0]; + return { + id: row.id as string, + name: row.name as string, + createdAt: new Date(row.created_at as string), + updatedAt: new Date(row.updated_at as string), + }; +}; + +export const findBucketByName = async (name: string): Promise => { + const result = (await db.execute( + sql`SELECT id, name, created_at, updated_at FROM buckets WHERE name = ${name}`, + )) as unknown as QueryResult; + if (result.rows.length === 0) return null; + const row = result.rows[0]; + return { + id: row.id as string, + name: row.name as string, + createdAt: new Date(row.created_at as string), + updatedAt: new Date(row.updated_at as string), + }; +}; + +export const listBuckets = async (): Promise => { + const result = (await db.execute( + sql`SELECT id, name, created_at, updated_at FROM buckets ORDER BY name`, + )) as unknown as QueryResult; + return result.rows.map((row) => ({ + id: row.id as string, + name: row.name as string, + createdAt: new Date(row.created_at as string), + updatedAt: new Date(row.updated_at as string), + })); +}; + +export const deleteBucket = async (name: string): Promise => { + const result = (await db.execute( + sql`DELETE FROM buckets WHERE name = ${name}`, + )) as unknown as QueryResult; + return result.rowCount > 0; +}; + +export const bucketExists = async (name: string): Promise => { + const result = (await db.execute( + sql`SELECT 1 FROM buckets WHERE name = ${name}`, + )) as unknown as QueryResult; + return result.rows.length > 0; +}; diff --git a/src/db/files-ext.ts b/src/db/files-ext.ts new file mode 100644 index 0000000..13c39ef --- /dev/null +++ b/src/db/files-ext.ts @@ -0,0 +1,118 @@ +import { eq, and, sql } from 'drizzle-orm'; +import { db, files as fileSchema } from './index'; +import type { File } from './schema'; + +interface QueryResult { + rows: Record[]; + rowCount: number; +} + +export interface S3FileRecord extends File { + bucketId: string; + s3Key: string; +} + +export const findFileByBucketAndKey = async ( + bucketId: string, + s3Key: string, +): Promise => { + const result = await db + .select() + .from(fileSchema) + .where( + and( + eq(fileSchema.bucketId, bucketId), + eq(fileSchema.s3Key, s3Key), + eq(fileSchema.isDeleted, false), + ), + ) + .limit(1); + return result[0] || null; +}; + +export const listObjectsByPrefix = async ( + bucketId: string, + prefix: string, + delimiter: string | null, + maxKeys: number, + startAfter: string | null, +): Promise<{ objects: S3FileRecord[]; prefixes: string[] }> => { + // Use raw SQL for the complex prefix/startAfter query + let query = sql`SELECT * FROM files WHERE bucket_id = ${bucketId}::uuid AND is_deleted = false AND s3_key LIKE ${prefix + '%'}`; + + if (startAfter) { + query = sql`${query} AND s3_key > ${startAfter}`; + } + + query = sql`${query} ORDER BY s3_key LIMIT ${maxKeys + 1}`; + + const result = (await db.execute(query)) as unknown as QueryResult; + + if (delimiter === '/') { + const prefixSet = new Set(); + const objects: S3FileRecord[] = []; + + for (const row of result.rows) { + const s3Key = row.s3_key as string; + const relativeKey = s3Key.substring(prefix.length); + const slashIndex = relativeKey.indexOf('/'); + if (slashIndex >= 0) { + // It's under a subfolder — extract the folder prefix + const folderPrefix = prefix + relativeKey.substring(0, slashIndex + 1); + if (folderPrefix !== prefix) { + prefixSet.add(folderPrefix); + } + } else { + // It's a direct child object + objects.push({ + ...row, + bucketId: row.bucket_id as string, + s3Key: s3Key, + } as unknown as S3FileRecord); + } + } + + return { + objects: objects.slice(0, maxKeys), + prefixes: Array.from(prefixSet).sort(), + }; + } + + return { + objects: result.rows.slice(0, maxKeys).map( + (row) => + ({ + ...row, + bucketId: row.bucket_id as string, + s3Key: row.s3_key as string, + }) as unknown as S3FileRecord, + ), + prefixes: [], + }; +}; + +export const softDeleteFile = async (bucketId: string, s3Key: string): Promise => { + const result = (await db.execute( + sql`UPDATE files SET is_deleted = true WHERE bucket_id = ${bucketId}::uuid AND s3_key = ${s3Key} RETURNING id`, + )) as unknown as QueryResult; + return result.rows.length > 0; +}; + +export const softDeleteFilesBatch = async ( + bucketId: string, + keys: string[], +): Promise => { + let deleted = 0; + for (const key of keys) { + const ok = await softDeleteFile(bucketId, key); + if (ok) deleted++; + } + return deleted; +}; + +export const countBucketObjects = async (bucketId: string): Promise => { + const result = (await db.execute( + sql`SELECT count(*) as count FROM files WHERE bucket_id = ${bucketId}::uuid AND is_deleted = false`, + )) as unknown as QueryResult; + return Number(result.rows[0]?.count || 0); +}; diff --git a/src/db/multipart.ts b/src/db/multipart.ts new file mode 100644 index 0000000..938acff --- /dev/null +++ b/src/db/multipart.ts @@ -0,0 +1,97 @@ +import { sql } from 'drizzle-orm'; +import { db } from './index'; +import { nanoid } from 'nanoid'; + +export interface MultipartUpload { + uploadId: string; + bucketId: string; + s3Key: string; + initiatedAt: Date; + status: string; + initiatedBy: string; +} + +export interface MultipartPart { + id: number; + uploadId: string; + partNumber: number; + telegramFileId: string; + telegramFileUniqueId: string; + storageMessageId: number; + sizeBytes: number; + etag: string; + createdAt: Date; +} + +interface QueryResult { + rows: Record[]; + rowCount: number; +} + +export const createMultipartUpload = async ( + bucketId: string, + s3Key: string, + initiatedBy: string, +): Promise => { + const uploadId = nanoid(32); + await db.execute( + sql`INSERT INTO multipart_uploads (upload_id, bucket_id, s3_key, initiated_by) VALUES (${uploadId}, ${bucketId}, ${s3Key}, ${initiatedBy})`, + ); + return uploadId; +}; + +export const findMultipartUpload = async (uploadId: string): Promise => { + const result = (await db.execute( + sql`SELECT upload_id, bucket_id, s3_key, initiated_at, status FROM multipart_uploads WHERE upload_id = ${uploadId} AND status = 'in_progress'`, + )) as unknown as QueryResult; + if (result.rows.length === 0) return null; + const r = result.rows[0]; + return { + uploadId: r.upload_id as string, + bucketId: r.bucket_id as string, + s3Key: r.s3_key as string, + initiatedAt: new Date(r.initiated_at as string), + status: r.status as string, + initiatedBy: '', + }; +}; + +export const completeMultipartUpload = async (uploadId: string): Promise => { + await db.execute( + sql`UPDATE multipart_uploads SET status = 'completed' WHERE upload_id = ${uploadId}`, + ); +}; + +export const abortMultipartUpload = async (uploadId: string): Promise => { + await db.execute( + sql`UPDATE multipart_uploads SET status = 'aborted' WHERE upload_id = ${uploadId}`, + ); + // Parts are cascade-deleted by FK +}; + +export const insertMultipartPart = async ( + part: Omit, +): Promise => { + await db.execute( + sql`INSERT INTO multipart_parts (upload_id, part_number, telegram_file_id, telegram_file_unique_id, storage_message_id, size_bytes, etag) + VALUES (${part.uploadId}, ${part.partNumber}, ${part.telegramFileId}, ${part.telegramFileUniqueId}, ${part.storageMessageId}, ${part.sizeBytes}, ${part.etag})`, + ); +}; + +export const listMultipartParts = async (uploadId: string): Promise => { + const result = (await db.execute( + sql`SELECT id, upload_id, part_number, telegram_file_id, telegram_file_unique_id, storage_message_id, size_bytes, etag, created_at + FROM multipart_parts WHERE upload_id = ${uploadId} ORDER BY part_number`, + )) as unknown as QueryResult; + return result.rows.map((r) => ({ + id: r.id as number, + uploadId: r.upload_id as string, + partNumber: r.part_number as number, + telegramFileId: r.telegram_file_id as string, + telegramFileUniqueId: r.telegram_file_unique_id as string, + storageMessageId: r.storage_message_id as number, + sizeBytes: r.size_bytes as number, + etag: r.etag as string, + createdAt: new Date(r.created_at as string), + })); +}; diff --git a/src/db/schema.ts b/src/db/schema.ts index 1a1d1a5..fa05941 100644 --- a/src/db/schema.ts +++ b/src/db/schema.ts @@ -1,5 +1,5 @@ import type { InferInsertModel, InferSelectModel } from 'drizzle-orm'; -import { bigint, pgTable, text, timestamp, uuid } from 'drizzle-orm/pg-core'; +import { bigint, boolean, pgTable, text, timestamp, uuid } from 'drizzle-orm/pg-core'; export const files = pgTable('files', { id: uuid('id').primaryKey().defaultRandom(), @@ -20,6 +20,11 @@ export const files = pgTable('files', { archiveEntryName: text('archive_entry_name'), archiveMimeType: text('archive_mime_type'), archiveSizeBytes: bigint('archive_size_bytes', { mode: 'number' }), + bucketId: text('bucket_id'), + s3Key: text('s3_key'), + storageBackend: text('storage_backend').default('telegram'), + isDeleted: boolean('is_deleted').default(false), + multipartUploadId: text('multipart_upload_id'), createdAt: timestamp('created_at').defaultNow().notNull(), updatedAt: timestamp('updated_at').defaultNow().notNull(), });