Use visible NAS upload storage and reconcile missing files

apps/server/src/accounts.ts

@@ -401,6 +401,7 @@ export class Accounts {
       )
       .get();
     if (!row) fail(404, 'ITEM_NOT_FOUND', '자료가 없거나 만료되었습니다.');
+    if (row!.item.fileId) this.service.storage.availability.require(row!.item.fileId);
     return row!;
   }
   read(userId: string, id: string, value: boolean) {
@@ -425,6 +426,7 @@ export class Accounts {
     query: { cursor?: string; limit: number; search?: string; kind?: string; read?: boolean },
   ) {
     this.verified(userId);
+    this.service.storage.availability.reconcile({ id: userId, kind: 'user' });
     let after: { createdAt: number; id: string } | undefined;
     if (query.cursor) {
       try {
@@ -520,6 +522,7 @@ export class Accounts {
     };
   }
   vaultFiles(userId: string) {
+    this.service.storage.availability.reconcile({ id: userId, kind: 'user' });
     const rows = this.service.db
       .select({ file: s.files })
       .from(s.access)

apps/server/src/app.ts

@@ -6,7 +6,6 @@ class PrivateLogs extends LogController {
 }
 import cookie from '@fastify/cookie';
 import multipart from '@fastify/multipart';
-import { createReadStream } from 'node:fs';
 import serveStatic from '@fastify/static';
 import { Server } from 'socket.io';
 import { timingSafeEqual } from 'node:crypto';
@@ -602,7 +601,7 @@ export async function createApp(cfg: Config, options: { mailer?: Mailer } = {})
   app.get('/api/access/:id/download', async (req, reply) => {
     const accessId = id(req),
       g = accessSubject(req, accessId);
-    const { file, path } = service.storage.download(g, accessId);
+    const { file, stream } = await service.storage.openDownload(g, accessId);
     const fallback =
       file.name.replace(/[^a-zA-Z0-9._ -]/g, '_').replace(/^[. ]+/, '') || 'download';
     const encoded = encodeURIComponent(file.name).replace(
@@ -616,8 +615,13 @@ export async function createApp(cfg: Config, options: { mailer?: Mailer } = {})
       )
       .header('Content-Length', file.size)
       .type('application/octet-stream');
-    readAccess(g, accessId, true);
-    return reply.send(createReadStream(path));
+    try {
+      readAccess(g, accessId, true);
+    } catch (error) {
+      stream.destroy();
+      throw error;
+    }
+    return reply.send(stream);
   });
   app.get('/api/access/:id/preview', async (req, reply) =>
     reply

apps/server/src/fileAvailability.js

@@ -0,0 +1,115 @@
+import { statSync } from 'node:fs';
+import { resolve } from 'node:path';
+import { and, eq, gt, inArray } from 'drizzle-orm';
+import * as s from './db/schema.js';
+import { fail } from './security.js';
+
+export class FileAvailability {
+  constructor(service, directory) {
+    this.service = service;
+    this.directory = directory;
+    this.pending = new Set();
+  }
+  check(file) {
+    try {
+      if (statSync(resolve(this.directory, file.storageKey)).isFile()) return true;
+    } catch (error) {
+      if (!['ENOENT', 'ENOTDIR'].includes(error.code))
+        fail(
+          503,
+          'STORAGE_UNAVAILABLE',
+          '파일 저장소에 접근할 수 없습니다. 잠시 후 다시 시도해 주세요.',
+        );
+    }
+    this.missing(file.id);
+    return false;
+  }
+  require(id) {
+    const file = this.service.db.select().from(s.files).where(eq(s.files.id, id)).get();
+    if (!file || file.status !== 'stored' || !this.check(file))
+      fail(404, 'FILE_NOT_FOUND', '서버의 파일이 삭제되었거나 만료되었습니다. 다시 전송해 주세요.');
+    return file;
+  }
+  reconcile(subject) {
+    const rows = this.service.db
+      .select({ file: s.files })
+      .from(s.files)
+      .innerJoin(s.items, eq(s.items.fileId, s.files.id))
+      .innerJoin(s.access, eq(s.access.itemId, s.items.id))
+      .where(
+        and(
+          eq(s.files.status, 'stored'),
+          subject.kind === 'user'
+            ? eq(s.access.userId, subject.id)
+            : eq(s.access.guestId, subject.id),
+        ),
+      )
+      .all();
+    for (const file of new Map(rows.map(({ file }) => [file.id, file])).values()) this.check(file);
+  }
+  invalidate(id) {
+    const db = this.service.db;
+    return db.transaction(() => {
+      const refs = db
+        .select({ access: s.access })
+        .from(s.access)
+        .innerJoin(s.items, eq(s.access.itemId, s.items.id))
+        .where(eq(s.items.fileId, id))
+        .all();
+      const transfers = db
+        .select({ id: s.requests.id })
+        .from(s.vaultTransfers)
+        .innerJoin(s.access, eq(s.vaultTransfers.accessId, s.access.id))
+        .innerJoin(s.items, eq(s.access.itemId, s.items.id))
+        .innerJoin(s.requests, eq(s.vaultTransfers.requestId, s.requests.id))
+        .innerJoin(s.batches, eq(s.batches.requestId, s.requests.id))
+        .where(
+          and(
+            eq(s.items.fileId, id),
+            inArray(s.requests.status, ['pending', 'accepted']),
+            eq(s.batches.status, 'waiting'),
+          ),
+        )
+        .all();
+      const now = Date.now();
+      db.update(s.files)
+        .set({ status: 'failed', checksum: null, updatedAt: now })
+        .where(and(eq(s.files.id, id), eq(s.files.status, 'stored')))
+        .run();
+      db.update(s.items)
+        .set({ expiresAt: now })
+        .where(and(eq(s.items.fileId, id), gt(s.items.expiresAt, now)))
+        .run();
+      db.delete(s.reservations).where(eq(s.reservations.fileId, id)).run();
+      for (const transfer of transfers) {
+        db.update(s.requests)
+          .set({ status: 'cancelled' })
+          .where(eq(s.requests.id, transfer.id))
+          .run();
+        db.update(s.batches)
+          .set({ status: 'failed' })
+          .where(eq(s.batches.requestId, transfer.id))
+          .run();
+      }
+      return { refs, transfers };
+    });
+  }
+  missing(id) {
+    const affected = this.invalidate(id);
+    if (this.pending.has(id)) return;
+    this.pending.add(id);
+    // A failed idempotent operation can roll back its outer transaction. Reapply afterward.
+    queueMicrotask(() => {
+      this.pending.delete(id);
+      const file = this.service.db.select().from(s.files).where(eq(s.files.id, id)).get();
+      const retried = file?.status === 'stored' ? this.invalidate(id) : { refs: [], transfers: [] };
+      const refs = [...affected.refs, ...retried.refs].map(({ access }) => access);
+      for (const user of new Set(refs.map((access) => access.userId).filter(Boolean)))
+        this.service.accounts.notify(user);
+      for (const guest of new Set(refs.map((access) => access.guestId).filter(Boolean)))
+        this.service.presence.emit(guest, 'access:updated', {});
+      for (const transfer of [...affected.transfers, ...retried.transfers])
+        this.service.notifyTransfer(transfer.id);
+    });
+  }
+}

apps/server/src/service.ts

@@ -843,6 +843,7 @@ export class Service {
       }));
   }
   listItems(g: Guest, cursor?: string, limit = 30) {
+    this.storage.availability.reconcile(g);
     let after: { createdAt: number; id: string } | undefined;
     if (cursor) {
       try {
@@ -933,6 +934,7 @@ export class Service {
       )
       .get();
     if (!row) fail(404, 'ITEM_NOT_FOUND', '자료가 없거나 만료되었습니다.');
+    if (row!.item.fileId) this.storage.availability.require(row!.item.fileId);
     return row!;
   }
   read(g: Guest, id: string, value: boolean) {
@@ -950,6 +952,7 @@ export class Service {
       )
       .get();
     if (!row) fail(404, 'ITEM_NOT_FOUND', '자료가 없거나 만료되었습니다.');
+    if (row!.item.fileId) this.storage.availability.require(row!.item.fileId);
     this.db
       .update(s.access)
       .set({ read: value })

apps/server/src/storage.ts

@@ -1,6 +1,6 @@
 import { createHash, randomUUID } from 'node:crypto';
-import { mkdirSync, statfsSync, existsSync } from 'node:fs';
-import { createWriteStream, createReadStream } from 'node:fs';
+import { mkdirSync, statfsSync } from 'node:fs';
+import { createWriteStream } from 'node:fs';
 import { rename, unlink, readdir, stat, open } from 'node:fs/promises';
 import { resolve } from 'node:path';
 import { Transform, Writable } from 'node:stream';
@@ -11,6 +11,7 @@ import type { FastifyRequest } from 'fastify';
 import type { Guest, Service } from './service.js';
 import * as s from './db/schema.js';
 import { ApiError, fail } from './security.js';
+import { FileAvailability } from './fileAvailability.js';
 export type StorageSubject = { id: string; kind?: 'user'; sessionId?: string };
 export class Storage {
   private active = new Map<
@@ -21,11 +22,13 @@ export class Storage {
   private cleanupPromise: Promise<void> | undefined;
   readonly temp: string;
   readonly originals: string;
+  readonly availability: FileAvailability;
   constructor(readonly service: Service) {
     this.temp = resolve(service.cfg.uploadDir, 'temporary');
     this.originals = resolve(service.cfg.uploadDir, 'originals');
     mkdirSync(this.temp, { recursive: true, mode: 0o700 });
     mkdirSync(this.originals, { recursive: true, mode: 0o700 });
+    this.availability = new FileAvailability(service, this.originals);
   }
   private request(g: StorageSubject, id: string) {
     return g.kind === 'user'
@@ -72,6 +75,7 @@ export class Storage {
         .run();
   }
   private used(guestId?: string) {
+    if (guestId) this.availability.reconcile({ id: guestId });
     return this.service.db
       .select({ bytes: sql<number>`coalesce(sum(${s.files.size}),0)` })
       .from(s.files)
@@ -429,14 +433,34 @@ export class Storage {
     const row = this.access(g, accessId);
     if (row.item.kind !== 'file' || !row.item.fileId)
       fail(404, 'FILE_NOT_FOUND', '파일을 찾을 수 없습니다.');
-    const file = this.service.db
-      .select()
-      .from(s.files)
-      .where(and(eq(s.files.id, row.item.fileId!), eq(s.files.status, 'stored')))
-      .get();
-    if (!file || !existsSync(resolve(this.originals, file.storageKey)))
-      fail(404, 'FILE_NOT_FOUND', '파일을 찾을 수 없습니다.');
-    return { file: file!, path: resolve(this.originals, file!.storageKey) };
+    const file = this.availability.require(row.item.fileId!);
+    return { file, path: resolve(this.originals, file.storageKey) };
+  }
+  async openDownload(g: StorageSubject, accessId: string) {
+    const { file, path } = this.download(g, accessId);
+    try {
+      const handle = await open(path, 'r');
+      return { file, stream: handle.createReadStream() };
+    } catch (error) {
+      if (
+        error &&
+        typeof error === 'object' &&
+        'code' in error &&
+        ['ENOENT', 'ENOTDIR'].includes(String(error.code))
+      ) {
+        this.availability.missing(file.id);
+        throw new ApiError(
+          404,
+          'FILE_NOT_FOUND',
+          '서버의 파일이 삭제되었습니다. 다시 전송해 주세요.',
+        );
+      }
+      throw new ApiError(
+        503,
+        'STORAGE_UNAVAILABLE',
+        '파일 저장소에 접근할 수 없습니다. 잠시 후 다시 시도해 주세요.',
+      );
+    }
   }
   async preview(g: StorageSubject, id: string) {
     const { path } = this.download(g, id);
@@ -582,6 +606,7 @@ export class Storage {
         .where(inArray(s.files.status, ['stored', 'cleaning']))
         .all();
       for (const file of all) {
+        if (file.status === 'stored' && !this.availability.check(file)) continue;
         const refs = db
           .select({ id: s.items.id })
           .from(s.items)

apps/server/test/missing-files.test.js

@@ -0,0 +1,271 @@
+import { test } from 'node:test';
+import assert from 'node:assert/strict';
+import { randomUUID } from 'node:crypto';
+import { unlink, chmod } from 'node:fs/promises';
+import { unlinkSync } from 'node:fs';
+import { resolve } from 'node:path';
+import { eq } from 'drizzle-orm';
+import { fixture } from './helpers.js';
+import * as s from '../src/db/schema.js';
+
+async function signIn(f, client) {
+  const id = randomUUID(),
+    email = `${id}@hop.test`;
+  f.service.db
+    .insert(s.users)
+    .values({
+      id,
+      email,
+      displayName: email,
+      passwordHash: await f.service.auth.password('1234'),
+      status: 'active',
+      verifiedAt: Date.now(),
+      createdAt: Date.now(),
+    })
+    .run();
+  const response = await f.call(client, '/auth/login', 'POST', { email, password: '1234' });
+  assert.equal(response.statusCode, 200, response.body);
+  client.cookie += '; ' + String(response.headers['set-cookie']).split(';')[0];
+  client.socket.disconnect();
+  await f.connect(client);
+}
+async function save(f, client, bytes = Buffer.from('test')) {
+  const request = await f.call(
+    client,
+    '/vault',
+    'POST',
+    {
+      kind: 'file',
+      files: [{ name: 'saved.txt', size: bytes.length, type: 'text/plain' }],
+    },
+    randomUUID(),
+  );
+  assert.equal(request.statusCode, 200, request.body);
+  const boundary = 'missing-file-test';
+  const body = Buffer.concat([
+    Buffer.from(
+      `--${boundary}\r\nContent-Disposition: form-data; name="file"; filename="saved.txt"\r\nContent-Type: text/plain\r\n\r\n`,
+    ),
+    bytes,
+    Buffer.from(`\r\n--${boundary}--\r\n`),
+  ]);
+  const response = await f.call(
+    client,
+    `/uploads?requestId=${request.json().requestId}&fileIndex=0`,
+    'POST',
+    body,
+    randomUUID(),
+    { 'content-type': `multipart/form-data; boundary=${boundary}` },
+  );
+  assert.equal(response.statusCode, 200, response.body);
+  const row = f.service.storage.fileRows(request.json().requestId)[0];
+  const item = (await f.call(client, '/vault'))
+    .json()
+    .items.find((item) => item.file?.id === row.id);
+  return { file: row, item, path: resolve(f.service.storage.originals, row.storageKey) };
+}
+async function forward(f, sender, receiver, item, accept = true) {
+  const response = await f.call(
+    sender,
+    `/vault/${item.accessId}/send`,
+    'POST',
+    { targetDeviceId: receiver.deviceId },
+    randomUUID(),
+  );
+  assert.equal(response.statusCode, 200, response.body);
+  const id = response.json().requestId;
+  if (accept)
+    assert.equal(
+      (await f.call(receiver, `/transfer-requests/${id}/accept`, 'POST', {}, randomUUID()))
+        .statusCode,
+      200,
+    );
+  return id;
+}
+function limits(f) {
+  f.cfg.sessionRate = 60;
+  f.cfg.pairRate = 60;
+  f.cfg.ipRate = 100;
+}
+
+for (const trigger of ['download', 'list', 'quota', 'cleanup']) {
+  test(`수동 삭제(${trigger}): 공유 원본의 모든 기록·개인 용량 정리와 정상 파일/링크·페이지 유지`, async () => {
+    const f = await fixture();
+    try {
+      limits(f);
+      const a = await f.create(),
+        b = await f.create(),
+        outsider = await f.create();
+      await f.space(a, b);
+      await signIn(f, a);
+      await signIn(f, b);
+      const healthy = await save(f, a, Buffer.from('healthy'));
+      const link = await f.call(
+        a,
+        '/vault',
+        'POST',
+        { kind: 'url', urls: ['https://example.com/healthy'] },
+        randomUUID(),
+      );
+      assert.equal(link.statusCode, 200, link.body);
+      const missing = await save(f, a);
+      await forward(f, a, b, missing.item);
+      const received = (await f.call(b, '/inbox')).json().items[0];
+      assert.equal((await f.call(a, '/account/storage')).json().usedBytes, 11);
+      assert.equal((await f.call(b, '/account/storage')).json().usedBytes, 4);
+      await unlink(missing.path);
+      const foreign = await f.call(outsider, `/access/${missing.item.accessId}/download`);
+      assert(foreign.statusCode === 401 || foreign.statusCode === 404);
+      assert.equal(
+        f.service.db.select().from(s.files).where(eq(s.files.id, missing.file.id)).get().status,
+        'stored',
+        '무권한 접근으로 정리/존재 확인을 유발하지 않음',
+      );
+      if (trigger === 'download') {
+        const response = await f.call(b, `/access/${received.accessId}/download`);
+        assert.equal(response.statusCode, 404, response.body);
+        assert.match(response.json().message, /다시 전송/);
+      } else if (trigger === 'list') {
+        await f.call(a, '/vault');
+      } else if (trigger === 'quota') {
+        assert.equal((await f.call(b, '/account/storage')).json().usedBytes, 0);
+      } else {
+        const update = new Promise((resolve, reject) => {
+          const timer = setTimeout(() => reject(Error('missing notification')), 2000);
+          b.socket.once('inbox:updated', () => {
+            clearTimeout(timer);
+            resolve();
+          });
+        });
+        await f.service.storage.cleanup();
+        await update;
+      }
+      assert.equal((await f.call(b, '/inbox')).json().items.length, 0);
+      assert.equal((await f.call(b, '/device/items')).json().items.length, 0);
+      assert.equal(
+        (await f.call(a, '/vault/contents'))
+          .json()
+          .items.some((i) => i.file?.id === missing.file.id),
+        false,
+      );
+      assert.equal((await f.call(a, '/account/storage')).json().usedBytes, 7);
+      assert.equal((await f.call(b, '/account/storage')).json().usedBytes, 0);
+      const page = (await f.call(a, '/vault?limit=1')).json();
+      assert.equal(page.items.length, 1);
+      assert(page.nextCursor);
+      const next = (
+        await f.call(a, `/vault?limit=1&cursor=${encodeURIComponent(page.nextCursor)}`)
+      ).json();
+      assert.equal(next.items.length, 1);
+      assert.equal(next.nextCursor, null);
+      const remaining = [...page.items, ...next.items];
+      assert(remaining.some((i) => i.file?.id === healthy.file.id));
+      assert(remaining.some((i) => i.url === 'https://example.com/healthy'));
+      assert.deepEqual(
+        (await f.call(a, `/access/${healthy.item.accessId}/download`)).rawPayload,
+        Buffer.from('healthy'),
+      );
+      assert.equal((await f.call(a, `/access/${missing.item.accessId}/preview`)).statusCode, 404);
+      f.service.cleanup();
+      await f.service.storage.cleanup();
+      assert.equal((await f.call(a, '/vault')).json().items.length, 2);
+    } finally {
+      await f.close();
+    }
+  });
+}
+
+for (const action of ['send', 'accept']) {
+  test(`수동 삭제 뒤 ${action} 실패: 외부 멱등 트랜잭션 롤백 후에도 삭제 감지와 대기 전송 취소 유지`, async () => {
+    const f = await fixture();
+    try {
+      limits(f);
+      const a = await f.create(),
+        b = await f.create();
+      await f.space(a, b);
+      await signIn(f, a);
+      const missing = await save(f, a);
+      const pending = action === 'accept' ? await forward(f, a, b, missing.item, false) : null;
+      await unlink(missing.path);
+      const result =
+        action === 'accept'
+          ? await f.call(b, `/transfer-requests/${pending}/accept`, 'POST', {}, randomUUID())
+          : await f.call(
+              a,
+              `/vault/${missing.item.accessId}/send`,
+              'POST',
+              { targetDeviceId: b.deviceId },
+              randomUUID(),
+            );
+      assert.equal(result.statusCode, 404, result.body);
+      assert.equal(
+        f.service.db.select().from(s.files).where(eq(s.files.id, missing.file.id)).get().status,
+        'failed',
+      );
+      assert.equal((await f.call(a, '/account/storage')).json().usedBytes, 0);
+      assert.equal((await f.call(a, '/vault')).json().items.length, 0);
+      assert.equal((await f.call(b, '/device/items')).json().items.length, 0);
+      if (pending) {
+        assert.equal(
+          f.service.db.select().from(s.requests).where(eq(s.requests.id, pending)).get().status,
+          'cancelled',
+        );
+        assert.equal(
+          f.service.db.select().from(s.batches).where(eq(s.batches.requestId, pending)).get()
+            .status,
+          'failed',
+        );
+      }
+    } finally {
+      await f.close();
+    }
+  });
+}
+
+test('권한 오류는 삭제로 오인하지 않고 503 안내 후 권한 복구 시 원본·DB·용량 유지', async (t) => {
+  if (process.getuid?.() === 0) return t.skip('root bypasses filesystem permissions');
+  const f = await fixture();
+  try {
+    const a = await f.create();
+    await signIn(f, a);
+    const saved = await save(f, a);
+    await chmod(f.service.storage.originals, 0);
+    const response = await f.call(a, `/access/${saved.item.accessId}/download`);
+    assert.equal(response.statusCode, 503, response.body);
+    assert.equal(response.json().code, 'STORAGE_UNAVAILABLE');
+    assert.equal(
+      f.service.db.select().from(s.files).where(eq(s.files.id, saved.file.id)).get().status,
+      'stored',
+    );
+    await chmod(f.service.storage.originals, 0o700);
+    assert.equal((await f.call(a, '/account/storage')).json().usedBytes, 4);
+    assert.deepEqual(
+      (await f.call(a, `/access/${saved.item.accessId}/download`)).rawPayload,
+      Buffer.from('test'),
+    );
+  } finally {
+    await chmod(f.service.storage.originals, 0o700);
+    await f.close();
+  }
+});
+
+test('메타데이터 확인 직후 원본이 사라져도 다운로드는 500 대신 명확한 404 반환', async () => {
+  const f = await fixture();
+  try {
+    const a = await f.create();
+    await signIn(f, a);
+    const saved = await save(f, a);
+    const download = f.service.storage.download.bind(f.service.storage);
+    f.service.storage.download = (...args) => {
+      const result = download(...args);
+      unlinkSync(saved.path);
+      return result;
+    };
+    const response = await f.call(a, `/access/${saved.item.accessId}/download`);
+    assert.equal(response.statusCode, 404, response.body);
+    assert.equal(response.json().code, 'FILE_NOT_FOUND');
+    assert.equal((await f.call(a, '/account/storage')).json().usedBytes, 0);
+  } finally {
+    await f.close();
+  }
+});