Worker 설정

// sonamu.config.ts
import { defineConfig } from "sonamu";

export default defineConfig({
  // ...
  tasks: {
    enableWorker: true, // Worker 활성화
    workerOptions: {
      concurrency: 4, // 동시 실행 수
      usePubSub: true, // Pub/Sub 사용
      listenDelay: 500, // 수신 지연 (ms)
    },
    contextProvider: async (baseContext) => {
      // Workflow에서 사용할 Context 설정
      return {
        ...baseContext,
        // 커스텀 데이터 추가 가능
      };
    },
  },
});

import { Sonamu } from "sonamu";

// Workflow 실행
const workflowManager = Sonamu.workflows;
const handle = await workflowManager.run(
  {
    name: "send_email",
    version: null,
  },
  { email: "user@example.com" },
);

// 결과 대기
const result = await handle.result();

# Worker가 자동으로 시작됨
pnpm sonamu daemon

// src/worker.ts
import { Sonamu } from "sonamu";

async function startWorker() {
  await Sonamu.init(true, false);

  await Sonamu.workflows.startWorker();

  console.log("Worker started");

  // Graceful shutdown
  const shutdown = async () => {
    console.log("Stopping worker...");
    await Sonamu.workflows.stopWorker();
    process.exit(0);
  };

  process.on("SIGTERM", shutdown);
  process.on("SIGINT", shutdown);
}

startWorker();

// package.json
{
  "scripts": {
    "worker": "tsx src/worker.ts",
    "worker:dev": "tsx watch src/worker.ts"
  }
}

# Worker 실행
pnpm worker

// sonamu.config.ts
export default defineConfig({
  tasks: {
    enableWorker: true, // API 서버에서도 Worker 실행
    workerOptions: {
      concurrency: 2, // API 서버는 적은 concurrency
    },
    contextProvider: async (baseContext) => baseContext,
  },
});

// Server 1, 2, 3...
tasks: {
  enableWorker: true,
  workerOptions: {
    concurrency: 4,
    usePubSub: true,
  },
}

// sonamu.config.ts
tasks: {
  enableWorker: true,
  contextProvider: async (baseContext) => {
    // 시스템 사용자 조회
    const systemUser = await UserModel.findOne({
      wq: [['email', '=', 'system@example.com']],
    });

    return {
      ...baseContext,
      user: systemUser,
      customData: "...",
    };
  },
}

import { workflow } from "sonamu";
import { Sonamu } from "sonamu";

export const myWorkflow = workflow({ name: "my_workflow" }, async ({ input, step }) => {
  const context = Sonamu.getContext();
  console.log(context.user); // systemUser
});

-- Workflow 정의
CREATE TABLE workflows (
  id UUID PRIMARY KEY,
  name TEXT NOT NULL,
  version TEXT,
  ...
);

-- Workflow 실행 기록
CREATE TABLE workflow_runs (
  id UUID PRIMARY KEY,
  workflow_id UUID REFERENCES workflows(id),
  status TEXT,
  input JSONB,
  output JSONB,
  started_at TIMESTAMP,
  completed_at TIMESTAMP,
  ...
);

-- Step 실행 기록
CREATE TABLE workflow_steps (
  id UUID PRIMARY KEY,
  workflow_run_id UUID REFERENCES workflow_runs(id),
  name TEXT,
  status TEXT,
  ...
);

// Puri를 통해 조회 (workflow_runs는 entity가 아닌 프레임워크 내부 테이블이므로 .knex 이스케이프 해치 사용)
import { BaseModel } from "sonamu";

const puri = BaseModel.getPuri("r");
const runs = await puri.knex.table("workflow_runs").where("status", "running").select("*");

export const myWorkflow = workflow({ name: "my_workflow" }, async ({ logger, input }) => {
  logger.info("Workflow started", { input });
  logger.error("Error occurred", { error });
});