EmProps Job Queue

Appearance

EmProps REST API ​

The EmProps API is the main platform interface for managing collections, workflows, jobs, and user data. It integrates with the Job Queue system to execute AI generation workloads.

Base URL ​

Production: https://api.emprops.comDevelopment: http://localhost:3001

Location: /Users/the_dusky/code/emerge/emerge-turbo/apps/emprops-api

Authentication ​

User ID Header ​

http
user_id: <uuid>

Required for all user-scoped operations. Set by authentication middleware from JWT tokens or session data.

API Key (Service-to-Service) ​

http
X-API-Key: <service-api-key>

Used for internal service communication and third-party integrations.

Collection Generation ​

Generate Collection ​

Submit a collection for AI generation.

Endpoint: POST /api/collections/:id/generate (or legacy /generator/v2/:id)

Request:

typescript
{
  variables: Record<string, any>
  workflow_id?: string
  workflow_priority?: number  // 1-100, default 50
}

Response:

typescript
{
  data: {
    jobId: string  // Job ID for tracking
  },
  error: null
}

Implementation: /apps/emprops-api/src/routes/generator/v2.ts

Process Flow:

  1. Validates collection data structure
  2. Creates job record in PostgreSQL
  3. Initializes GeneratorV2 with event handlers
  4. Submits workflow steps to Job Queue (Redis)
  5. Returns immediately with job ID

Event Handlers:

  • node_started → Updates job status to "processing"
  • node_progress → Updates job progress percentage
  • node_completed → Records step completion
  • complete → Marks job as completed with workflow_output
  • error → Marks job as failed with error message

Get Job Status ​

Retrieve current status of a generation job.

Endpoint: GET /api/jobs/:id

Response:

typescript
{
  data: {
    id: string
    name: string
    description: string
    status: "pending" | "processing" | "completed" | "failed"
    progress: number | null  // 0-100
    error_message: string | null
    created_at: string
    updated_at: string
    started_at: string | null
    completed_at: string | null
    job_type: string
    priority: number
    workflow_output: string | null  // Final output URL
    data: {
      collectionId: string
      variables: Record<string, any>
      outputs?: any
    }
  },
  error: null
}

Stream Job Events (SSE) ​

Real-time job updates via Server-Sent Events.

Endpoint: GET /api/jobs/:id/stream

Response (Event Stream):

event: job_update
data: {"id":"...","status":"processing","progress":50,...}

event: job_history
data: {"id":"...","status":"processing","message":"Node started",...}

event: history_init
data: [{"id":"...","status":"pending",...},{...}]

:heartbeat

Note: Prisma 6 removed middleware support. WebSocket job updates are currently disabled pending migration to $extends() API.

Retry Job ​

Retry a failed or stuck job.

Endpoint: POST /api/jobs/:id/retry

Response:

typescript
{
  data: {
    jobId: string  // Same job ID, incremented retry_count
    retryAttempt: number
    message: string
  },
  error: null
}

Retry Logic:

  • Preserves job ID, increments retry_count
  • Backs up original data to job_retry_backup table
  • Resets job status to "pending"
  • Re-executes GeneratorV2 with same workflow
  • Captures workflow_output directly (no extraction needed)

Workflow Management ​

List Workflows ​

Get available workflows with optional filters.

Endpoint: GET /api/workflows

Query Parameters:

  • include=eq.true - Include full workflow data
  • require_api_keys=true - Filter workflows requiring API keys
  • Standard pagination parameters

Response:

typescript
{
  data: [
    {
      id: string
      name: string
      label: string
      type: "basic" | "comfy_workflow" | "fetch_api" | "direct_job" | "dynamic_json"
      description: string
      created_at: string
      data?: any  // Included if include=eq.true
    }
  ],
  error: null
}

Get Workflow by Name ​

Retrieve workflow with dependencies.

Endpoint: GET /api/workflows/:name

Response:

typescript
{
  data: {
    id: string
    name: string
    label: string
    type: string
    description: string
    data: any
    models: string[]  // Required model names
    custom_nodes: string[]  // Required custom node names
  },
  error: null
}

Create Workflow ​

Create a new workflow definition.

Endpoint: POST /api/workflows

Request:

typescript
{
  name: string
  label: string
  description: string
  data: any
  output_mime_type: string
  type?: "basic" | "comfy_workflow" | "fetch_api" | "direct_job" | "dynamic_json"
  display?: boolean
  order?: number
  models?: string[]  // Model names to link
  custom_nodes?: string[]  // Custom node names to link
}

Custodial Collections (API-First) ​

Create Simple Collection ​

Create a custodial collection with dynamic JSON templates.

Endpoint: POST /api/workflows/create-simple-collection

Request:

typescript
{
  collection_details: {
    title: string
    description: string
    social_org: "farcaster" | "twitter"
    social_identifier: string
    blockchain?: "tezos" | "base"  // default: "tezos"
    price?: string | number  // default: "0"
    editions?: number | string  // default: 1
    cover_image_url?: string
    miniapp_user_id?: string
    miniapp_cover_image?: string
    cast_hash?: string
    project_id?: string  // Will use user's default if not provided
  },
  variables: [
    {
      name: string
      template_mapping: string  // Maps to {{variable}} in template
      options?: string[]
      is_selectable?: boolean
      is_customizable?: boolean
    }
  ],
  job_type: string  // e.g., "openai_chat", "stability_api"
  template_json: any  // JSON template with {{variables}}
}

Response:

typescript
{
  data: {
    id: string
    title: string
    description: string
    blockchain: string
    price: number
    editions: number
    status: string
    is_custodial: boolean
    custodied_for: string  // social_link_id
    social_link: {
      id: string
      social_org: string
      social_identifier: string
      miniapp_user_id: string | null
    },
    variables: [...],
    component: {
      type: "dynamic_json"
      job_type: string
      template: any
    }
  },
  error: null
}

Implementation Details:

  • Creates social_link if doesn't exist
  • Converts variables to prompt-editor format with embedded references
  • Uses flexible_prompt workflow (type: direct_job)
  • Template variables are replaced during generation

File Upload ​

Upload Base64 ​

Upload a file from base64-encoded data to cloud storage. Optionally creates a flat_file database record for asset management.

Endpoint: POST /upload-base64

Authentication: Authorization: Bearer <token> or X-Api-Key

Request:

typescript
{
  base64Data: string          // Base64-encoded file content (required)
  filename?: string           // Original filename (default: auto-generated UUID)
  mimeType?: string           // MIME type (default: "image/png")
  create_flat_file?: boolean  // Create a flat_file DB record (default: false)
  rel_type?: string           // flat_file rel_type (default: "upload", e.g. "nft_ref_image")
  miniapp_user_id?: string    // UUID — preferred user identifier
  fid?: string                // Farcaster ID — fallback identifier
}

Headers:

  • user_id — User UUID (used for storage path and flat_file ownership when miniapp_user_id/fid not provided)

Response:

typescript
{
  data: {
    url: string           // Storage path (e.g. "user-uploads/user-id/{userId}/{uuid}.png")
    path: string          // Same as url
    size: number          // File size in bytes
    mimeType: string      // MIME type
    flatFileId?: number   // flat_file record ID (only when create_flat_file=true)
  },
  error: null
}

Storage Path Structure:

user-uploads/{pathType}/{userId}/{uuid}.{ext}

Examples:
  user-uploads/user-id/abc-123/def456.png          (from user_id header)
  user-uploads/miniapp-userid/abc-123/def456.png    (from miniapp_user_id)
  user-uploads/fid/12345/def456.png                 (from Farcaster ID)

Limits:

  • Maximum file size: 10MB
  • Allowed MIME types: image/png, image/jpeg, image/webp, image/gif, video/mp4, video/webm

Implementation: /apps/emprops-api/src/routes/upload-base64.ts

NFT Samples ​

Generate Samples ​

Generate sample images for a collection with variable overrides.

Endpoint: POST /nft/collections/:collectionId/samples

Request:

typescript
{
  count: number            // 1-10
  wallet_address: string   // 0x-prefixed address
  variables?: Record<string, any>  // Variable overrides (customizable = any value, selectable = preset value)
  tx_hash?: string         // USDC payment hash (required for paid samples)
  chain_id?: number        // Blockchain network ID
}

Response:

typescript
{
  success: true,
  samples: [
    { outputId: string, jobId: string, status: "pending", is_free: boolean }
  ]
}

List Samples ​

Endpoint: GET /nft/collections/:collectionId/samples

Response:

typescript
{
  success: true,
  samples: [
    {
      outputId: string
      url: string | null
      status: "completed" | "pending" | "failed"
      is_free: boolean
      created_at: string
      jobId: string | null
      traits: Array<{ trait_type: string, value: string }>
    }
  ]
}

Sample Status ​

Endpoint: GET /nft/collections/:collectionId/sample-status

Response:

typescript
{
  success: true,
  free_remaining: number
  free_limit_per_collection: number
  paid_count: number
  compute_cost_per_sample: number
}

Implementation: /apps/emprops-api/src/routes/nft/samples.ts

Model & Node Management ​

List Models ​

Get available AI models.

Endpoint: GET /api/models

List Custom Nodes ​

Get ComfyUI custom nodes.

Endpoint: GET /api/custom-nodes

Failsafe Endpoints ​

Failsafe Job Complete ​

Notify job completion when webhook fails.

Endpoint: POST /api/jobs/:id/complete

Request:

typescript
{
  outputs: any
  metadata?: any
  workflow_output: string  // Final output URL (required)
}

Purpose: Prevents infinite webhook loops when job completes but EmProps doesn't receive notification.

Failsafe Job Failed ​

Notify job failure when webhook fails.

Endpoint: POST /api/jobs/:id/failed

Request:

typescript
{
  error_message: string
  error_details?: any
}

Step Tracking (Internal) ​

Create Step ​

Create step record before WebSocket submission.

Function: createStep(prisma, stepId, jobId, stepName, stepType, inputData?, stepOrder?, retryAttempt?)

Location: /apps/emprops-api/src/routes/jobs/index.ts

Update Step Complete ​

Mark step as completed with retry logic.

Function: updateStepComplete(prisma, stepId, outputData?, retries = 3)

Features:

  • Exponential backoff (1s, 2s, 4s)
  • Verification after update
  • Automatic retry on transient failures

Reconcile Stale Steps ​

Background process to fix stuck steps using Redis attestation.

Function: reconcileStaleSteps(prisma)

Logic:

  1. Finds steps pending >10 minutes
  2. Queries Redis for authoritative status
  3. Updates PostgreSQL based on Redis truth
  4. Handles: completed, failed, still processing

Error Responses ​

Collection Corrupted ​

typescript
{
  data: null,
  error: "COLLECTION_CORRUPTED",
  message: "This collection contains corrupted data...",
  userMessage: "Collection data is corrupted. Please contact support.",
  details: {...}
}

Insufficient Credits ​

typescript
{
  data: null,
  error: "Not enough credits."
}

Job Retry Limit ​

typescript
{
  data: null,
  error: "Job has reached maximum retry limit (3)"
}

Integration with Job Queue ​

How EmProps API Submits Jobs ​

  1. Create Job Record (PostgreSQL)

    typescript
    await prisma.job.create({
  data: {
    id: jobId,
    name: "Collection Generation: ...",
    status: "pending",
    data: { collectionId, variables },
    user_id: userId,
    job_type: "collection_generation",
    priority: workflowPriority
  }
})

  2. Submit Steps to Job Queue (Redis)

    typescript
    // GeneratorV2 submits each workflow step
for (const step of workflow.steps) {
  await submitJobToQueue({
    id: stepId,
    type: "comfyui" | "openai_chat" | ...,
    payload: stepPayload,
    requirements: {
      models: [...],
      custom_nodes: [...]
    },
    workflow_context: {
      workflow_id: jobId,
      workflow_priority: priority,
      total_steps: workflow.steps.length,
      current_step: index + 1
    }
  })
}

  3. Receive Results (Webhooks/Redis)

    • Workers publish progress to Redis channels
    • Webhooks deliver final results to EmProps API
    • API updates job record with workflow_output

File Locations ​

  • Routes: /apps/emprops-api/src/routes/

    • generator/v2.ts - Collection generation
    • jobs/index.ts - Job management and step tracking
    • workflows/index.ts - Workflow CRUD
    • collections/metadata.ts - Collection metadata

  • Core Logic: /apps/emprops-api/src/modules/art-gen/nodes-v2/

    • generator.ts - GeneratorV2 orchestrator
    • nodes/ - Workflow node implementations

  • Database: /apps/emprops-api/prisma/schema.prisma

See Also ​

Released under the MIT License.

Copyright © 2024 EmProps