workflow-studio/compute-worker

packages/compute-worker/
├── src/
│   ├── index.ts           # Main server implementation
│   ├── index.test.ts      # Server tests
│   └── types.ts           # TypeScript interfaces
└── package.json

CREATE TABLE IF NOT EXISTS api_keys (
  id TEXT PRIMARY KEY,
  project_id TEXT NOT NULL,
  environment TEXT NOT NULL,
  scopes TEXT NOT NULL,  -- JSON array of scopes
  secret_hash TEXT NOT NULL,  -- SHA-256 of secret
  revoked_at TEXT,
  created_at TEXT NOT NULL
);

export type ComputeWorkerOptions = {
  seedApiKeys?: SeedApiKey[];
  // ...
};

type SeedApiKey = {
  keyId: string;
  projectId: string;
  environment: string;
  scopes: ComputeScope[];
  secret: string;
};

CREATE TABLE IF NOT EXISTS idempotency (
  project_id TEXT NOT NULL,
  idempotency_key TEXT NOT NULL,
  endpoint TEXT NOT NULL,
  status_code INTEGER NOT NULL,
  body TEXT NOT NULL,
  created_at INTEGER NOT NULL,
  state TEXT,             -- 'in_progress' | 'completed' (used by /v1/runs)
  request_hash TEXT,      -- Canonical JSON hash for /v1/runs payload
  response_run_id TEXT,   -- Stable run ID across retries
  PRIMARY KEY (project_id, idempotency_key, endpoint)
);

CREATE TABLE IF NOT EXISTS deployments (
  deployment_id TEXT PRIMARY KEY,
  manifest_json TEXT NOT NULL,
  artifact_path TEXT NOT NULL,
  status TEXT NOT NULL,  -- 'created', 'active', 'inactive'
  created_at TEXT NOT NULL,
  activated_at TEXT
);

CREATE TABLE IF NOT EXISTS settings (
  key TEXT PRIMARY KEY,
  value TEXT NOT NULL
);

-- Active deployment pointer
-- key = 'active_deployment', value = deployment_id

CREATE TABLE IF NOT EXISTS audit_logs (
  id TEXT PRIMARY KEY,
  key_id TEXT,
  project_id TEXT,
  action TEXT NOT NULL,
  resource TEXT NOT NULL,
  result TEXT NOT NULL,  -- 'success' or 'failure'
  ip TEXT,
  user_agent TEXT,
  metadata_json TEXT,  -- JSON object with lane, deduped, etc.
  created_at TEXT NOT NULL
);

function authorize(
  db: SqliteDB,
  req: Request,
  requiredScope: ComputeScope
): AuthContext | Response {
  const token = readBearer(req.headers);
  if (!token) {
    return json({ code: 'unauthorized', message: 'Missing API key' }, { status: 401 });
  }
  
  const row = verifyToken(db, token);
  if (!row) {
    return json({ code: 'unauthorized', message: 'Invalid API key' }, { status: 401 });
  }
  
  const auth: AuthContext = {
    keyId: row.id,
    projectId: row.projectId,
    scopes: new Set(JSON.parse(row.scopes) as ComputeScope[]),
  };
  
  const scopeError = ensureScope(auth, requiredScope);
  if (scopeError) return scopeError;
  
  return auth;
}

function verifyToken(db: SqliteDB, secret: string): ApiKeyRecord | null {
  const rows = db.prepare(/* ... */).all() as ApiKeyRecord[];
  const provided = Buffer.from(sha256(secret));
  
  for (const row of rows) {
    const stored = Buffer.from(row.secretHash);
    if (stored.length === provided.length && timingSafeEqual(stored, provided)) {
      return row;
    }
  }
  return null;
}

GET /v1/health
Authorization: Bearer <token>  # optional

Response: { healthy: true, timestamp: "2024-01-01T00:00:00.000Z" }

GET /v1/deployments/active
Authorization: Bearer <token>  # requires 'deploy:read' scope

Response: { deploymentId: "dep_abc123" }
## or 409 { code: "no_active_deployment" }

POST /v1/runs
Authorization: Bearer <token>  # requires 'trigger:write' scope
Idempotency-Key: <key>  # required
Content-Type: application/json

Body: {
  workflowName: "send-email",
  input?: {...},
  runId?: "wrun_...",  // optional, auto-generated if omitted
  deploymentId?: "dep_...",  // optional, uses active if omitted
  specVersion?: 1
}

Response: { 
  runId: "wrun_...", 
  status: "pending", 
  deploymentId: "dep_..." 
}

POST /v1/runs/resume
Authorization: Bearer <token>  # requires 'runs:write' scope
Content-Type: application/json

Body: {
  runId: "wrun_...",
  reason?: "manual-retry",
  payload?: {...}
}

Response: {
  accepted: true
}

GET /v1/world/deployment-id
Authorization: Bearer <token>  # requires 'world:proxy' scope

Response: { deploymentId: "dep_abc123" }
## or 409 { code: "no_active_deployment" }

POST /v1/queue/publish
Authorization: Bearer <token>  # requires 'world:proxy' scope
Content-Type: application/json

Body: {
  queueName: "__wkf_workflow_send-email",
  message: { runId: "wrun_..." },
  opts?: {
    deploymentId?: "dep_...",
    idempotencyKey?: "key-123"  // explicit only
  }
}

Response: { messageId: "msg_..." }

{
  "code": "invalid_deployment_id",
  "message": "deploymentId must match ^[A-Za-z0-9_-]+$ and cannot contain path separators."
}

function enforceRemoteTargetGuard(env: NodeJS.ProcessEnv): void {
  const mode = executionMode ?? env.WORKFLOW_EXECUTION_MODE;
  if (mode !== 'remote' || !shouldApplyStrictRemoteGuard(env)) {
    return;
  }
  
  const expectedTarget = env.WORKFLOW_STUDIO_REMOTE_WORLD_TARGET ?? 'workflow-studio/world-remote';
  const actualTarget = env.WORKFLOW_TARGET_WORLD;
  
  if (actualTarget !== expectedTarget) {
    throw new Error(
      `Remote execution guard failed: WORKFLOW_TARGET_WORLD must equal "${expectedTarget}"...`
    );
  }
}

function applyRateLimit(
  db: SqliteDB,
  auth: AuthContext,
  limit: number  // default 120/minute
): Response | undefined {
  const windowStart = Math.floor(Date.now() / 60_000) * 60_000;
  
  // Upsert count
  db.prepare(`
    INSERT INTO rate_limits (key_id, project_id, window_start, count)
    VALUES (?, ?, ?, 1)
    ON CONFLICT(key_id, project_id, window_start)
    DO UPDATE SET count = count + 1
  `).run(auth.keyId, auth.projectId, windowStart);
  
  // Check limit
  const row = db.prepare(/* select count */).get(/* ... */);
  if ((row?.count ?? 0) > limit) {
    return json({ code: 'rate_limited', message: 'Rate limit exceeded' }, { status: 429 });
  }
}

function enforceHttps(req: Request, requireHttps: boolean): Response | undefined {
  if (!requireHttps) return undefined;
  
  const protocol = req.headers.get('x-forwarded-proto');
  if (protocol && protocol !== 'https') {
    return json(
      { code: 'https_required', message: 'HTTPS is required' },
      { status: 400 }
    );
  }
  return undefined;
}