hyb-oa/server/src/services/storage/index.js

const fs = require("node:fs/promises");
const fsSync = require("node:fs");
const path = require("node:path");

const { S3Client, PutObjectCommand, GetObjectCommand } = require("@aws-sdk/client-s3");

class StorageNotFoundError extends Error {
  constructor(message) {
    super(message);
    this.name = "StorageNotFoundError";
  }
}

class StorageConfigError extends Error {
  constructor(message) {
    super(message);
    this.name = "StorageConfigError";
  }
}

function parseStoragePath(storagePath) {
  const raw = typeof storagePath === "string" ? storagePath : "";
  if (raw.startsWith("s3:")) {
    return { driver: "s3", key: raw.slice("s3:".length) };
  }
  return { driver: "local", key: raw };
}

function ensureSafeLocalPath(storageRoot, relativeKey) {
  const root = path.resolve(storageRoot);
  const absolutePath = path.resolve(root, relativeKey);
  if (!absolutePath.startsWith(`${root}${path.sep}`)) {
    throw new StorageConfigError("Invalid storage_path (path traversal detected)");
  }
  return absolutePath;
}

function resolveS3Config(config) {
  if (!config?.s3Bucket) {
    throw new StorageConfigError("Missing S3_BUCKET for s3 storage");
  }

  return {
    bucket: config.s3Bucket,
    region: config.s3Region || "us-east-1",
    endpoint: config.s3Endpoint || undefined,
    credentials:
      config.s3AccessKeyId && config.s3SecretAccessKey
        ? {
            accessKeyId: config.s3AccessKeyId,
            secretAccessKey: config.s3SecretAccessKey,
          }
        : undefined,
    forcePathStyle: Boolean(config.s3ForcePathStyle),
  };
}

let cachedS3Client = null;
let cachedS3ClientKey = null;

function getS3Client(config) {
  const s3 = resolveS3Config(config);
  const cacheKey = JSON.stringify({
    bucket: s3.bucket,
    region: s3.region,
    endpoint: s3.endpoint ?? "",
    accessKeyId: s3.credentials?.accessKeyId ?? "",
    forcePathStyle: s3.forcePathStyle,
  });

  if (cachedS3Client && cacheKey === cachedS3ClientKey) {
    return cachedS3Client;
  }

  cachedS3ClientKey = cacheKey;
  cachedS3Client = new S3Client({
    region: s3.region,
    endpoint: s3.endpoint,
    credentials: s3.credentials,
    forcePathStyle: s3.forcePathStyle,
  });
  return cachedS3Client;
}

async function streamToBuffer(stream) {
  const chunks = [];
  for await (const chunk of stream) {
    chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk));
  }
  return Buffer.concat(chunks);
}

async function writeBuffer({ config, key, buffer, contentType }) {
  if (!Buffer.isBuffer(buffer) || buffer.length === 0) {
    throw new StorageConfigError("Invalid buffer");
  }

  const driver = config?.storageDriver ?? "local";
  if (driver === "s3") {
    const s3 = resolveS3Config(config);
    const client = getS3Client(config);

    await client.send(
      new PutObjectCommand({
        Bucket: s3.bucket,
        Key: key,
        Body: buffer,
        ContentType: contentType || "application/octet-stream",
      }),
    );

    // Use a lightweight prefix to support mixed local+s3 records without schema changes.
    return `s3:${key}`;
  }

  const absolutePath = ensureSafeLocalPath(config.storageDir, key);
  await fs.mkdir(path.dirname(absolutePath), { recursive: true });
  await fs.writeFile(absolutePath, buffer);
  return key;
}

async function readFileBuffer({ config, storagePath }) {
  const parsed = parseStoragePath(storagePath);
  if (!parsed.key) {
    throw new StorageNotFoundError("missing_storage_path");
  }

  if (parsed.driver === "s3") {
    const s3 = resolveS3Config(config);
    const client = getS3Client(config);

    try {
      const result = await client.send(
        new GetObjectCommand({
          Bucket: s3.bucket,
          Key: parsed.key,
        }),
      );
      if (!result?.Body) {
        throw new StorageNotFoundError("s3_body_missing");
      }
      return streamToBuffer(result.Body);
    } catch (error) {
      const code = error?.$metadata?.httpStatusCode;
      if (code === 404 || error?.name === "NoSuchKey") {
        throw new StorageNotFoundError("s3_not_found");
      }
      throw error;
    }
  }

  const absolutePath = ensureSafeLocalPath(config.storageDir, parsed.key);
  try {
    return await fs.readFile(absolutePath);
  } catch (error) {
    if (error?.code === "ENOENT") {
      throw new StorageNotFoundError("local_not_found");
    }
    throw error;
  }
}

async function openReadStream({ config, storagePath }) {
  const parsed = parseStoragePath(storagePath);
  if (!parsed.key) {
    throw new StorageNotFoundError("missing_storage_path");
  }

  if (parsed.driver === "s3") {
    const s3 = resolveS3Config(config);
    const client = getS3Client(config);

    try {
      const result = await client.send(
        new GetObjectCommand({
          Bucket: s3.bucket,
          Key: parsed.key,
        }),
      );

      if (!result?.Body) {
        throw new StorageNotFoundError("s3_body_missing");
      }

      return { stream: result.Body };
    } catch (error) {
      const code = error?.$metadata?.httpStatusCode;
      if (code === 404 || error?.name === "NoSuchKey") {
        throw new StorageNotFoundError("s3_not_found");
      }
      throw error;
    }
  }

  const absolutePath = ensureSafeLocalPath(config.storageDir, parsed.key);
  try {
    await fs.access(absolutePath);
  } catch (error) {
    if (error?.code === "ENOENT") {
      throw new StorageNotFoundError("local_not_found");
    }
    throw error;
  }

  return { stream: fsSync.createReadStream(absolutePath) };
}

module.exports = {
  StorageConfigError,
  StorageNotFoundError,
  openReadStream,
  parseStoragePath,
  readFileBuffer,
  writeBuffer,
};