Task - 🌲 Sonamu

Sonamu는 비동기 백그라운드 작업을 처리하는 Workflow 시스템을 제공합니다. 긴 작업을 백그라운드로 분리하여 API 응답 속도를 개선하고, Worker 프로세스로 작업을 분산 처리할 수 있습니다.

기본 구조

import { defineConfig } from "sonamu";

export default defineConfig({
  tasks: {
    enableWorker: true,
    workerOptions: {
      concurrency: 1,
      usePubSub: true,
      listenDelay: 500,
    },
    contextProvider: (defaultContext) => ({
      ...defaultContext,
      ip: "127.0.0.1",
      session: {},
    }),
  },
  // ...
});

enableWorker

Worker 프로세스를 활성화할지 결정합니다. 타입: boolean (선택적) 기본값: false

export default defineConfig({
  tasks: {
    enableWorker: true, // Worker 활성화
  },
});

Worker란?

별도 프로세스에서 작업을 처리
메인 API 서버에 영향 없음
여러 Worker를 실행하여 분산 처리 가능

프로덕션 환경에서는 enableWorker: true로 설정하여 Worker 전용 프로세스를 실행하는 것을 권장합니다.

환경 변수로 제어

export default defineConfig({
  tasks: {
    enableWorker: !["true", "1"].includes(process.env.DISABLE_WORKER ?? "false"),
  },
});

.env:

# Worker 비활성화 (개발)
DISABLE_WORKER=true

# Worker 활성화 (프로덕션)
DISABLE_WORKER=false

workerOptions

Worker의 동작 방식을 설정합니다. 타입: WorkflowOptions (선택적)

type WorkflowOptions = {
  concurrency?: number;
  usePubSub?: boolean;
  listenDelay?: number;
};

concurrency

Worker가 동시에 처리할 작업 수입니다. 타입: number 기본값: 1

export default defineConfig({
  tasks: {
    workerOptions: {
      concurrency: 1, // 한 번에 1개 작업만
    },
  },
});

권장값:

1 - 단순한 작업, 순차 처리 필요
2-5 - 일반적인 백그라운드 작업
10+ - 가벼운 I/O 위주 작업

concurrency를 높이면 더 많은 작업을 동시에 처리하지만, CPU와 메모리 사용량이 증가합니다.

usePubSub

다중 Worker 간 작업 분산을 위해 Pub/Sub을 사용합니다. 타입: boolean 기본값: true

export default defineConfig({
  tasks: {
    workerOptions: {
      usePubSub: true, // PostgreSQL LISTEN/NOTIFY 사용
    },
  },
});

동작 방식:

false: 각 Worker가 독립적으로 작업 풀링
true: PostgreSQL LISTEN/NOTIFY로 새 작업 알림을 즉시 전파

Pub/Sub은 PostgreSQL의 내장 LISTEN/NOTIFY 메커니즘을 사용하므로, 별도의 인프라(Redis 등) 설치가 필요하지 않습니다. 이미 구성된 데이터베이스 연결을 그대로 활용합니다.

listenDelay

작업을 확인하는 주기(밀리초)입니다. 타입: number (ms) 기본값: 500 (0.5초)

export default defineConfig({
  tasks: {
    workerOptions: {
      listenDelay: 500, // 0.5초마다 작업 확인
    },
  },
});

권장값:

100-500ms - 빠른 응답이 필요한 작업
500-1000ms - 일반적인 백그라운드 작업
1000-5000ms - 긴급하지 않은 작업

contextProvider

Worker에서 실행되는 작업의 Context를 생성합니다. 타입: (defaultContext) => Context | Promise<Context>

export default defineConfig({
  tasks: {
    contextProvider: (defaultContext) => {
      return {
        ...defaultContext,
        ip: "127.0.0.1",
        session: {},
      };
    },
  },
});

defaultContext 포함 필드:

reply - Fastify reply 객체 (null in worker)
request - Fastify request 객체 (null in worker)
headers - 요청 헤더
createSSE - SSE 스트림 생성
naiteStore - Naite 로깅 스토어

커스텀 Context

export default defineConfig({
  tasks: {
    contextProvider: (defaultContext) => {
      return {
        ...defaultContext,
        ip: "worker",
        session: {},
        config: {
          apiUrl: process.env.API_URL,
        },
        logger: console,
      };
    },
  },
});

Worker Context에서는 request와 reply가 null입니다. HTTP 관련 기능은 사용할 수 없습니다.

기본 예시

단일 Worker (개발)

import { defineConfig } from "sonamu";

export default defineConfig({
  tasks: {
    enableWorker: process.env.NODE_ENV === "production",
    workerOptions: {
      concurrency: 1,
      usePubSub: false, // 단일 Worker
      listenDelay: 500,
    },
    contextProvider: (defaultContext) => ({
      ...defaultContext,
      ip: "127.0.0.1",
      session: {},
    }),
  },
});

다중 Worker (프로덕션)

import { defineConfig } from "sonamu";

export default defineConfig({
  tasks: {
    enableWorker: true,
    workerOptions: {
      concurrency: 5,
      usePubSub: true, // PostgreSQL LISTEN/NOTIFY로 분산
      listenDelay: 500,
    },
    contextProvider: (defaultContext) => ({
      ...defaultContext,
      ip: "worker",
      session: {},
    }),
  },
});

실전 예시

환경별 설정

import { defineConfig } from "sonamu";

const isDev = process.env.NODE_ENV === "development";
const isProd = process.env.NODE_ENV === "production";

export default defineConfig({
  tasks: {
    // 개발: Worker 없음 (메인 프로세스에서 처리)
    // 프로덕션: Worker 활성화
    enableWorker: isProd,

    workerOptions: {
      concurrency: isProd ? 5 : 1,
      usePubSub: isProd,
      listenDelay: isDev ? 100 : 500, // 개발: 빠른 응답
    },

    contextProvider: (defaultContext) => ({
      ...defaultContext,
      ip: isProd ? "worker" : "127.0.0.1",
      session: {},
    }),
  },
});

고부하 워크플로우

export default defineConfig({
  tasks: {
    enableWorker: true,
    workerOptions: {
      concurrency: 10, // 동시에 10개 작업
      usePubSub: true, // 다중 Worker 지원
      listenDelay: 1000, // 1초 간격
    },
    contextProvider: (defaultContext) => ({
      ...defaultContext,
      ip: "worker",
      session: {},
      logger: {
        info: (msg) => console.log(`[Worker] ${msg}`),
        error: (msg) => console.error(`[Worker Error] ${msg}`),
      },
    }),
  },
});

커스텀 Context

export default defineConfig({
  tasks: {
    enableWorker: true,
    workerOptions: {
      concurrency: 3,
      usePubSub: true,
      listenDelay: 500,
    },
    contextProvider: async (defaultContext) => {
      // 비동기 초기화
      const config = await loadConfig();

      return {
        ...defaultContext,
        ip: "worker",
        session: {},
        config,
        services: {
          email: new EmailService(),
          notification: new NotificationService(),
        },
      };
    },
  },
});

Workflow 사용

tasks 설정 후 workflow() 함수로 백그라운드 작업을 정의합니다.

import { workflow } from "sonamu";

export const sendWelcomeEmail = workflow(
  { name: "send_welcome_email" },
  async ({ input, step }) => {
    const user = await step
      .define({ name: "fetch_user" }, async () => {
        return await UserModel.findById(input.userId);
      })
      .run();

    await step
      .define({ name: "send_email" }, async () => {
        await sendEmail(user.email, "Welcome!");
      })
      .run();
  },
);

호출:

import { Sonamu } from "sonamu";

// 백그라운드에서 실행
await Sonamu.workflows.run({ name: "send_welcome_email", version: null }, { userId: 123 });

→ Workflow 사용법

Worker 실행

개발 환경

개발 환경에서는 enableWorker: false로 두고 메인 프로세스에서 처리:

pnpm dev

프로덕션 환경

Worker 전용 프로세스를 별도로 실행:

# API 서버
pnpm start

# Worker 프로세스 (별도 터미널)
pnpm start:worker

package.json:

{
  "scripts": {
    "start": "node dist/server.js",
    "start:worker": "ENABLE_WORKER=true node dist/worker.js"
  }
}

다중 Worker

여러 Worker를 실행하여 처리량 증가:

# Worker 1
ENABLE_WORKER=true node dist/worker.js

# Worker 2
ENABLE_WORKER=true node dist/worker.js

# Worker 3
ENABLE_WORKER=true node dist/worker.js

PM2나 Docker를 사용하면 여러 Worker를 쉽게 관리할 수 있습니다.

Pub/Sub 동작 방식

usePubSub: true(기본값)일 때, Worker는 PostgreSQL의 내장 LISTEN/NOTIFY 메커니즘을 사용하여 새 작업 알림을 수신합니다. 별도의 메시지 브로커(Redis 등)가 필요하지 않으며, 이미 구성된 데이터베이스 연결을 그대로 활용합니다.

export default defineConfig({
  tasks: {
    enableWorker: true,
    workerOptions: {
      usePubSub: true, // PostgreSQL LISTEN/NOTIFY 사용 (기본값)
    },
  },
});

주의사항

1. Worker Context 제한

// ❌ 나쁜 예: Worker에서 HTTP 응답 시도
export const processData = workflow({ name: "process_data" }, async () => {
  const ctx = Sonamu.getContext();
  ctx.reply.send({ done: true }); // Worker에서는 null!
});

// ✅ 좋은 예: 결과를 DB에 저장
export const processData = workflow({ name: "process_data" }, async () => {
  const result = await heavyProcessing();
  await ResultModel.save({ data: result });
});

2. enableWorker 설정

// ❌ 나쁜 예: 프로덕션에서 Worker 없음
enableWorker: false; // 모든 작업이 메인 프로세스에서!

// ✅ 좋은 예: 환경별로 다르게
enableWorker: process.env.NODE_ENV === "production";

3. concurrency 과다 설정

// ❌ 나쁜 예: 너무 높은 동시성
workerOptions: {
  concurrency: 100,  // CPU와 메모리 부족 위험
}

// ✅ 좋은 예: 적절한 수준
workerOptions: {
  concurrency: 5,  // 서버 리소스에 맞게
}

다음 단계

Task 설정을 완료했다면:

workflows - Workflow 정의하기
steps - Step으로 작업 분리
error-handling - 에러 처리 및 재시도

​기본 구조

​enableWorker

​환경 변수로 제어

​workerOptions

​concurrency

​usePubSub

​listenDelay

​contextProvider

​커스텀 Context

​기본 예시

​단일 Worker (개발)

​다중 Worker (프로덕션)

​실전 예시

​환경별 설정

​고부하 워크플로우

​커스텀 Context

​Workflow 사용

​Worker 실행

​개발 환경

​프로덕션 환경

​다중 Worker

​Pub/Sub 동작 방식

​주의사항

​1. Worker Context 제한

​2. enableWorker 설정

​3. concurrency 과다 설정

​다음 단계

기본 구조

enableWorker

환경 변수로 제어

workerOptions

concurrency

usePubSub

listenDelay

contextProvider

커스텀 Context

기본 예시

단일 Worker (개발)

다중 Worker (프로덕션)

실전 예시

환경별 설정

고부하 워크플로우

커스텀 Context

Workflow 사용

Worker 실행

개발 환경

프로덕션 환경

다중 Worker

Pub/Sub 동작 방식

주의사항

1. Worker Context 제한

2. enableWorker 설정

3. concurrency 과다 설정

다음 단계