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

user_id: <uuid>

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

API Key (Service-to-Service)

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:

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

Response:

{
  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:

{
  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:

{
  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:

{
  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:

{
  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:

{
  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:

{
  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:

{
  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

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:

{
  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:

{
  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

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

Insufficient Credits

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

Job Retry Limit

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

Integration with Job Queue

How EmProps API Submits Jobs

1. Create Job Record (PostgreSQL)

   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)

   // 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

• WebSocket API - Real-time job queue events
• Internal APIs - Redis pub/sub and webhooks
• Architecture Overview - System design