job-manager

.github/workflows/docker-publish.yml

@@ -0,0 +1,44 @@
+name: Build and Push Docker Image
+
+on:
+  push:
+    branches: [ main-v2 ]
+    paths:
+      - 'job-manager/job-runner/**'
+  workflow_dispatch:
+
+env:
+  REGISTRY: ghcr.io
+  IMAGE_NAME: ${{ github.repository }}-job-runner
+
+jobs:
+  build-and-push:
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read
+      packages: write
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+
+      - name: Log in to the Container registry
+        uses: docker/login-action@v3
+        with:
+          registry: ${{ env.REGISTRY }}
+          username: ${{ github.actor }}
+          password: ${{ secrets.GITHUB_TOKEN }}
+
+      - name: Extract metadata (tags, labels) for Docker
+        id: meta
+        uses: docker/metadata-action@v5
+        with:
+          images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}
+
+      - name: Build and push Docker image
+        uses: docker/build-push-action@v5
+        with:
+          context: job-manager/job-runner
+          push: true
+          tags: ${{ steps.meta.outputs.tags }}
+          labels: ${{ steps.meta.outputs.labels }}

CHANGELOG.md

@@ -1,6 +1,7 @@
 # Changes
 
 ## February 18, 2025
+- Added distributed job processing system (neurosift-job-manager)
 - Added support for embedded=1 query parameter
 - Added feedback/issues link on home page footer
 - Migration to v2 deployed

job-manager/.env.example

@@ -0,0 +1,8 @@
+# MongoDB connection string
+MONGODB_URI=mongodb://localhost:27017/neurosift_jobs
+
+# API Keys for authentication
+API_SUBMIT_KEY=your-submit-key-here
+
+# Memobin API Key for file storage
+MEMOBIN_API_KEY=your-memobin-key-here

job-manager/.gitignore

@@ -0,0 +1,55 @@
+# dependencies
+/node_modules
+/.pnp
+.pnp.js
+
+# testing
+/coverage
+
+# next.js
+/.next/
+/out/
+/build/
+
+# misc
+.DS_Store
+*.pem
+
+# debug
+npm-debug.log*
+yarn-debug.log*
+yarn-error.log*
+
+# local env files
+.env
+.env*.local
+
+# vercel
+.vercel
+
+# typescript
+*.tsbuildinfo
+next-env.d.ts
+
+# Python
+__pycache__/
+*.py[cod]
+*$py.class
+*.so
+.Python
+venv/
+ENV/
+.env
+*.egg-info/
+dist/
+build/
+
+# IDE
+.idea/
+.vscode/
+*.swp
+*.swo
+
+# OS
+.DS_Store
+Thumbs.db

job-manager/README.md

@@ -0,0 +1,55 @@
+# Neurosift Job Manager
+
+A distributed job processing system that allows the Neurosift UI to submit processing jobs that are executed by remote compute clients.
+
+## Local Development Setup
+
+1. Prerequisites:
+   - Docker with Docker Compose plugin (not the standalone docker-compose)
+   - Node.js and npm
+   - Python 3.x
+
+   To install Docker Compose plugin:
+   ```bash
+   # For Debian/Ubuntu
+   sudo apt-get update
+   sudo apt-get install docker-compose-plugin
+
+   # For macOS with Homebrew
+   brew install docker-compose
+   ```
+
+2. Run the development setup script:
+   ```bash
+   ./dev-setup.sh
+   ```
+   This script will:
+   - Start MongoDB in a Docker container
+   - Create a `.env` file
+   - Install npm dependencies
+
+3. In a separate terminal, start the service:
+
+   ```bash
+   npm run dev
+   ```
+
+4. The following services will be available:
+   - MongoDB: mongodb://localhost:27017/neurosift_jobs
+   - API Server: http://localhost:3000/api
+
+To stop the services:
+1. Stop the API server  Ctrl+C in their terminal
+2. Stop MongoDB: `docker compose down`
+
+## Running a job
+
+1. Install Python dependencies:
+   ```bash
+   pip install -r requirements.txt
+   ```
+
+2. Start the compute client:
+   ```bash
+   ./run-job.py <JOB_ID>
+   ```

job-manager/app/api/jobs/[id]/route.ts

@@ -0,0 +1,146 @@
+/**
+ * API Routes for Individual Job Operations
+ *
+ * This module provides endpoints for retrieving and updating specific jobs
+ * using their unique identifier. Supports both job status checks and
+ * progress updates.
+ */
+
+import { NextRequest, NextResponse } from 'next/server';
+import connectDB, { Job, IJob } from '../../../../lib/db';
+import { validateJobState } from '../../../../middleware/auth';
+import { UpdateQuery } from 'mongoose';
+
+/**
+ * Handle CORS preflight requests
+ * Enables cross-origin requests from the development server
+ *
+ * @returns Response with appropriate CORS headers
+ */
+export async function OPTIONS() {
+  return new NextResponse(null, {
+    status: 200,
+    headers: {
+      'Access-Control-Allow-Origin': 'http://localhost:5173',
+      'Access-Control-Allow-Methods': 'GET, POST, PUT, DELETE, PATCH, OPTIONS',
+      'Access-Control-Allow-Headers': 'Content-Type, Authorization'
+    }
+  });
+}
+
+/**
+ * Retrieve details of a specific job
+ *
+ * @param request NextRequest - The incoming request
+ * @param params.id string - The unique identifier of the job
+ * @returns
+ *   - Success: Job object with all details
+ *   - Error: 404 if job not found, 500 for server errors
+ *
+ * No API key required - job ID is used as authentication
+ */
+export async function GET(
+  request: NextRequest,
+  { params }: { params: { id: string } }
+) {
+  try {
+    await connectDB();
+    const job = await Job.findById(params.id);
+
+    // Simple not found check is sufficient since job ID is considered secret
+    if (!job) {
+      return new NextResponse('Job not found', { status: 404 });
+    }
+
+    return NextResponse.json(job);
+  } catch (error) {
+    console.error('Error fetching job:', error);
+    return new NextResponse('Internal Server Error', { status: 500 });
+  }
+}
+
+/**
+ * Update the status and progress of a specific job
+ *
+ * @param request NextRequest containing any of:
+ *   - status: 'pending' | 'running' | 'completed' | 'failed'
+ *   - progress: number (0-100)
+ *   - output: job output data
+ *   - error: error information if job failed
+ * @param params.id string - The unique identifier of the job
+ * @returns
+ *   - Success: Updated job object
+ *   - Error: 404 if job not found, 500 for server errors
+ *
+ * Required API key: 'fulfill'
+ *
+ * Only specified fields will be updated, maintaining data integrity
+ */
+export async function PATCH(
+  request: NextRequest,
+  { params }: { params: { id: string } }
+) {
+  // No API key needed for job updates - we use job ID as authentication
+
+  try {
+    const body = await request.json();
+    const { status, progress, output, error } = body;
+
+    await connectDB();
+    const job = await Job.findById(params.id);
+
+    // Validate job exists and state transition is valid
+    const validationError = await validateJobState(job, status);
+    if (validationError) return validationError;
+
+    // Only allow updates to specific fields based on current state
+    const updates: UpdateQuery<IJob> = {};
+
+    if (status) {
+      updates.$set = updates.$set || {};
+      updates.$set.status = status;
+    }
+
+    // Only allow progress updates for running jobs
+    if (typeof progress === 'number') {
+      if (job.status !== 'running' && status !== 'running') {
+        return new NextResponse('Progress can only be updated for running jobs', { status: 400 });
+      }
+      if (progress < 0 || progress > 100) {
+        return new NextResponse('Progress must be between 0 and 100', { status: 400 });
+      }
+      updates.$set = updates.$set || {};
+      updates.$set.progress = progress;
+    }
+
+    // Only allow output updates for running or completed jobs
+    if (output) {
+      if (!['running', 'completed'].includes(job.status) &&
+          !(status && ['running', 'completed'].includes(status))) {
+        return new NextResponse('Output can only be set for running or completed jobs', { status: 400 });
+      }
+      updates.$set = updates.$set || {};
+      updates.$set.output = output;
+    }
+
+    // Only allow error updates when transitioning to failed status
+    if (error) {
+      if (status !== 'failed') {
+        return new NextResponse('Error can only be set when status is being set to failed', { status: 400 });
+      }
+      updates.$set = updates.$set || {};
+      updates.$set.error = error;
+    }
+
+    const updatedJob = await Job.findByIdAndUpdate(
+      params.id,
+      updates,
+      { new: true, runValidators: true }
+    );
+
+    return NextResponse.json(updatedJob);
+  } catch (error) {
+    console.error('Error updating job:', error);
+    return new NextResponse('Internal Server Error', { status: 500 });
+  }
+}