Elastic Workers (Beta)

./workers/<name>/
  index.mjs        # Entry point (Node.js)
  package.json     # Dependencies (npm install runs on instance boot)

{
  "workers": [
    { "name": "my-worker", "instance": "16core", "timeout": 3600 }
  ]
}

openkbs worker deploy <name>                    # Deploy worker code
openkbs worker run <name> [-d '{"key":"val"}'] [-i <tier>] [-c spot]  # Launch instance, run worker
openkbs worker debug <name> [-i <tier>] [-t <sec>]  # SSH into a debug instance (auto-terminates)
openkbs worker list                             # List deployed workers (tier, last deploy)
openkbs worker runs [name]                      # List recent runs WITH run ids (filter by worker name)
openkbs worker logs <runId|name>                # View run output. A worker name tails its LATEST run.
openkbs worker stop <runId>                     # Cancel and terminate
openkbs worker types                            # Show tiers and pricing
openkbs worker destroy <name>                   # Delete worker

openkbs worker run my-worker -i 64core -c spot

// functions/api/index.mjs
const projectId = process.env.OPENKBS_PROJECT_ID;
const apiKey = process.env.OPENKBS_API_KEY;

// Trigger worker
const res = await fetch(
  `https://project.openkbs.com/projects/${projectId}/worker/my-worker/run`,
  {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
      Authorization: `Bearer ${apiKey}`,
    },
    body: JSON.stringify({
      tier: '16core',
      timeout: 3600,
      env: { ORDER_ID: orderId, CUSTOMER_NAME: name },
    }),
  }
);
const { runId, status, cost } = await res.json();
// Save runId, respond to user with "processing..."

// workers/my-worker/index.mjs
import { S3Client, GetObjectCommand, PutObjectCommand, ListObjectsV2Command } from '@aws-sdk/client-s3';
import { createWriteStream, readFileSync, mkdirSync } from 'fs';
import { pipeline } from 'stream/promises';
import { execSync } from 'child_process';
import { join, basename } from 'path';
import pg from 'pg';

const s3 = new S3Client({ region: process.env.STORAGE_REGION });
const db = new pg.Pool({ connectionString: process.env.DATABASE_URL, ssl: { rejectUnauthorized: true } });
const bucket = process.env.STORAGE_BUCKET;
const scratch = process.env.SCRATCH;
const orderId = process.env.ORDER_ID;

mkdirSync(join(scratch, 'input'), { recursive: true });
mkdirSync(join(scratch, 'output'), { recursive: true });

// 1. Download input files from S3 to fast NVMe
const objects = await s3.send(new ListObjectsV2Command({
  Bucket: bucket, Prefix: `uploads/${orderId}/`,
}));
for (const obj of objects.Contents || []) {
  const res = await s3.send(new GetObjectCommand({ Bucket: bucket, Key: obj.Key }));
  await pipeline(res.Body, createWriteStream(join(scratch, 'input', basename(obj.Key))));
}

// 2. Process with FFmpeg (or any system tool)
execSync(`ffmpeg -i ${scratch}/input/raw.mp4 -c:v libx264 -preset slow ${scratch}/output/result.mp4`);

// 3. Upload result to S3
await s3.send(new PutObjectCommand({
  Bucket: bucket,
  Key: `processed/${orderId}/result.mp4`,
  Body: readFileSync(join(scratch, 'output', 'result.mp4')),
  ContentType: 'video/mp4',
}));

// 4. Update database
await db.query('UPDATE orders SET status=$1, completed_at=now() WHERE id=$2', ['completed', orderId]);

// 5. Send real-time notification via MQTT
await fetch(`https://project.openkbs.com/projects/${process.env.OPENKBS_PROJECT_ID}/mqtt/publish`, {
  method: 'POST',
  headers: {
    'Content-Type': 'application/json',
    Authorization: `Bearer ${process.env.OPENKBS_API_KEY}`,
  },
  body: JSON.stringify({
    channel: `order-${orderId}`,
    event: 'completed',
    data: { videoUrl: `processed/${orderId}/result.mp4` },
  }),
});

await db.end();
console.log('Done');

openkbs worker debug <name> [-i <tier>] [-t <seconds>]

openkbs worker debug my-worker -i 16core -t 1800
# → prints:  ssh -i <key> ec2-user@<ip>

ssh -i <key> -o StrictHostKeyChecking=no ec2-user@<ip>   # connect (retry once if refused)

# on the instance:
cd /opt/worker
source .env.sh        # load the exact worker env (OPENKBS_API_KEY, DATABASE_URL, SCRATCH, ...)
node index.mjs        # run the worker by hand — observe, edit, re-run, iterate

exit
openkbs worker stop <runId>   # terminate early (optional; it auto-kills at the timeout)

GET    /worker/types                                          # Public, no auth — list tiers
GET    /projects/{id}/worker                                  # List deployed workers
POST   /projects/{id}/worker/{name}                           # Deploy worker (body: {code: "base64zip"})
DELETE /projects/{id}/worker/{name}                           # Delete worker
POST   /projects/{id}/worker/{name}/run                       # Run worker (body: {tier?, timeout?, env?})
POST   /projects/{id}/worker/{name}/debug                     # Debug worker, SSH access (body: {tier?, timeout?, publicKey})
GET    /projects/{id}/worker/runs                             # List recent runs
GET    /projects/{id}/worker/run/{runId}                      # Get run status
POST   /projects/{id}/worker/run/{runId}/stop                 # Stop and terminate
GET    /projects/{id}/worker/run/{runId}/logs                 # Get run logs
POST   /projects/{id}/worker/run/{runId}/callback             # Internal: EC2 instance reports completion

{ "code": "<base64-encoded zip of worker directory>" }

{
  "tier": "8core",
  "timeout": 600,
  "env": { "CUSTOM_VAR": "value" }
}