feat: add DB CRUD layer for buckets, multipart, and S3 file extensions

2026-07-06 14:23:37 +07:00
parent e37fb00295
commit caef31c703
4 changed files with 288 additions and 1 deletions
@@ -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<string, unknown>[];
+  rowCount: number;
+}
+
+export const createBucket = async (name: string): Promise<Bucket> => {
+  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<Bucket | null> => {
+  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<Bucket[]> => {
+  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<boolean> => {
+  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<boolean> => {
+  const result = (await db.execute(
+    sql`SELECT 1 FROM buckets WHERE name = ${name}`,
+  )) as unknown as QueryResult;
+  return result.rows.length > 0;
+};
@@ -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<string, unknown>[];
+  rowCount: number;
+}
+
+export interface S3FileRecord extends File {
+  bucketId: string;
+  s3Key: string;
+}
+
+export const findFileByBucketAndKey = async (
+  bucketId: string,
+  s3Key: string,
+): Promise<File | null> => {
+  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<string>();
+    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<boolean> => {
+  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<number> => {
+  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<number> => {
+  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);
+};
@@ -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<string, unknown>[];
+  rowCount: number;
+}
+
+export const createMultipartUpload = async (
+  bucketId: string,
+  s3Key: string,
+  initiatedBy: string,
+): Promise<string> => {
+  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<MultipartUpload | null> => {
+  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<void> => {
+  await db.execute(
+    sql`UPDATE multipart_uploads SET status = 'completed' WHERE upload_id = ${uploadId}`,
+  );
+};
+
+export const abortMultipartUpload = async (uploadId: string): Promise<void> => {
+  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<MultipartPart, 'id' | 'createdAt'>,
+): Promise<void> => {
+  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<MultipartPart[]> => {
+  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),
+  }));
+};
@@ -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(),
 });