Worker Setup

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

export default defineConfig({
  // ...
  tasks: {
    enableWorker: true, // Enable Worker
    workerOptions: {
      concurrency: 4, // Concurrent execution count
      usePubSub: true, // Use Pub/Sub
      listenDelay: 500, // Listen delay (ms)
    },
    contextProvider: async (baseContext) => {
      // Configure Context used in Workflows
      return {
        ...baseContext,
        // Custom data can be added
      };
    },
  },
});

import { Sonamu } from "sonamu";

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

// Wait for result
const result = await handle.result();

# Worker starts automatically
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"
  }
}

# Run Worker
pnpm worker

// sonamu.config.ts
export default defineConfig({
  tasks: {
    enableWorker: true, // Run Worker in API server too
    workerOptions: {
      concurrency: 2, // Lower concurrency for API server
    },
    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) => {
    // Fetch system user
    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 definitions
CREATE TABLE workflows (
  id UUID PRIMARY KEY,
  name TEXT NOT NULL,
  version TEXT,
  ...
);

-- Workflow execution records
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 execution records
CREATE TABLE workflow_steps (
  id UUID PRIMARY KEY,
  workflow_run_id UUID REFERENCES workflow_runs(id),
  name TEXT,
  status TEXT,
  ...
);

// Query via Puri (workflow_runs is a framework-internal table, not an entity, so use the .knex escape hatch)
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 });
});