Constructive Jobs

Constructive Jobs

Background job infrastructure for the Constructive platform. Declaratively attach triggers to tables that enqueue jobs when rows change, processed by the Knative worker stack.

When to Apply

• Adding a background job that fires on row INSERT/UPDATE/DELETE
• Wiring a table to a Knative cloud function (e.g., send email on invite creation)
• Syncing data to external systems on change (e.g., Stripe sync on invoice update)
• Generating embeddings, sending notifications, auditing changes
• Scheduling recurring jobs (cron-style)
• Adding file/image embeddings to a storage table

Architecture

Table row change (INSERT/UPDATE/DELETE)
  --> PostgreSQL AFTER trigger (created by JobTrigger node)
    --> app_jobs.add_job(task_identifier, payload)
      --> knative-job-worker polls app_jobs.jobs
        --> POST ${KNATIVE_SERVICE_URL}/${task_identifier}
          --> Knative function handles the job

The database extension pgpm-database-jobs provides:

• app_jobs.jobs — queued/running jobs table
• app_jobs.scheduled_jobs — cron-style scheduled jobs table
• app_jobs.add_job() — enqueue a one-off job
• app_jobs.add_scheduled_job() — register a recurring job

The JobTrigger blueprint node automatically creates the PostgreSQL triggers that call app_jobs.add_job().

JobTrigger Blueprint Node

Add to a table's nodes[] in a blueprint definition to auto-create triggers:

{
  ref: 'invoices',
  table_name: 'invoices',
  nodes: [
    ...ORG_NODES,
    {
      $type: 'JobTrigger',
      data: {
        task_identifier: 'process_invoice',
      }
    },
  ],
  fields: [
    { name: 'amount', type: { name: 'numeric' }, is_required: true },
    { name: 'status', type: { name: 'text' }, default_value: { value: 'draft' } },
  ],
}

This creates INSERT and UPDATE triggers that enqueue a process_invoice job with { id: row.id } as the payload.

Configuration Reference

Constraints: conditions, condition_field, and watch_fields are mutually exclusive — only one can be specified per trigger.

Compound Conditions

The conditions parameter accepts a structured JSON syntax for complex WHEN clauses. Column types are resolved automatically from the PostgreSQL schema — values in JSON are cast to the correct type at generation time. This system is shared with EventTracker (see constructive-events) — both use the same build_condition_ast() function and conditionProperties schema.

Leaf condition:

{ field: 'status', op: '=', value: 'ready', row: 'NEW' }

Array shorthand (implicit AND):

conditions: [
  { field: 'status', op: '=', value: 'ready' },
  { field: 'status', op: '=', value: 'pending', row: 'OLD' },
  { field: 'mime_type', op: 'LIKE', value: 'image/%' },
]

Nested combinators (AND/OR/NOT):

conditions: {
  AND: [
    { field: 'status', op: '=', value: 'ready' },
    { OR: [
      { field: 'mime_type', op: 'LIKE', value: 'image/%' },
      { field: 'mime_type', op: 'LIKE', value: 'video/%' },
    ]},
    { NOT: { field: 'is_draft', op: '=', value: true } },
  ]
}

See references/common-patterns.md for full blueprint examples.

Payload Strategies

See references/payload-strategies.md for detailed examples of each strategy.

Common Patterns

See references/common-patterns.md for full blueprint examples of:

• Conditional triggers (watch_fields, condition_field, conditions)
• Compound conditions (status transitions, MIME type filtering)
• Delayed/debounced jobs (run_at_delay + job_key)
• Multiple triggers per table
• Email on invite, Stripe sync, audit trail, webhook dispatch

ProcessFileEmbedding Blueprint Node

Generic, MIME-scoped embedding node for file/storage tables. Composes SearchVector + JobTrigger + ProcessChunks internally. Supports two modes:

• Direct mode (default): whole-file to single vector (e.g., CLIP for images). No extraction config.
• Extract mode: file to text to chunks to per-chunk vectors. Enabled by providing extraction config.

Multiple instances can coexist on the same table with different MIME scopes, field names, and embedding strategies.

Direct Mode (single vector per file)

// Image embeddings via CLIP — one vector per image file
{
  ref: 'files',
  table_name: 'files',
  nodes: [
    ...STORAGE_NODES,
    { $type: 'ProcessFileEmbedding', data: {
      mime_patterns: ['image/%'],
      dimensions: 512,
      task_identifier: 'process_image_embedding',
    }},
  ],
}

Extract Mode (file to text to chunks to vectors)

// Document embeddings — extract text, chunk, embed each chunk
{
  ref: 'files',
  table_name: 'files',
  nodes: [
    ...STORAGE_NODES,
    { $type: 'ProcessFileEmbedding', data: {
      mime_patterns: ['application/pdf', 'text/%', 'application/vnd.openxmlformats-officedocument.*'],
      dimensions: 768,
      task_identifier: 'process_document_extraction',
      extraction: {
        text_field: 'extracted_text',
        metadata_field: 'extracted_metadata',
      },
      // chunks are enabled by default in extract mode
      chunks: {
        chunk_size: 1000,
        chunk_overlap: 200,
        chunk_strategy: 'paragraph',
      },
    }},
  ],
}

Multi-Modal: Multiple Pipelines on One Table

// Knowledge base — three embedding pipelines on one files table
{
  ref: 'files',
  table_name: 'files',
  nodes: [
    ...STORAGE_NODES,

    // Pipeline 1: CLIP visual embeddings for images
    { $type: 'ProcessFileEmbedding', data: {
      field_name: 'image_embedding',
      mime_patterns: ['image/%'],
      dimensions: 512,
      task_identifier: 'process_image_embedding',
    }},

    // Pipeline 2: Text extraction + chunked embeddings for documents
    { $type: 'ProcessFileEmbedding', data: {
      field_name: 'document_embedding',
      mime_patterns: ['application/pdf', 'text/%', 'application/vnd.openxmlformats-officedocument.*'],
      dimensions: 768,
      task_identifier: 'process_document_extraction',
      extraction: {
        text_field: 'extracted_text',
        metadata_field: 'extracted_metadata',
      },
    }},

    // Pipeline 3: Audio/video transcription + chunked embeddings
    { $type: 'ProcessFileEmbedding', data: {
      field_name: 'media_embedding',
      mime_patterns: ['audio/%', 'video/%'],
      dimensions: 768,
      task_identifier: 'process_media_transcription',
      extraction: {
        text_field: 'transcription_text',
        metadata_field: 'transcription_metadata',
      },
    }},
  ],
}

Configuration Reference

ProcessImageEmbedding Blueprint Node

Image-specific preset of ProcessFileEmbedding. Delegates entirely to ProcessFileEmbedding with image-oriented defaults.

// Minimal — uses all defaults (512d CLIP, image/%, process_image_embedding)
{
  ref: 'files',
  table_name: 'files',
  nodes: [
    ...STORAGE_NODES,
    { $type: 'ProcessImageEmbedding' },
  ],
}

Default overrides vs ProcessFileEmbedding:

All ProcessFileEmbedding parameters are accepted and forwarded through. You can use ProcessImageEmbedding with extraction to enable OCR-based text extraction from images.

ProcessChunks Blueprint Node

Standalone chunking node that creates a child chunks table for any parent table. Composed internally by ProcessFileEmbedding (enabled by default in extract mode), but can also be used standalone.

The chunks table gets:

• FK to parent (CASCADE delete)
• content text field
• chunk_index integer
• embedding vector(N) with HNSW index
• metadata jsonb
• RLS policies inherited from parent
• Optional job trigger for automatic chunking

Standalone Usage

// Add chunking to any table with text content
{
  ref: 'articles',
  table_name: 'articles',
  nodes: [
    'DataId',
    'DataTimestamps',
    { $type: 'ProcessChunks', data: {
      chunk_size: 1000,
      chunk_overlap: 200,
      chunk_strategy: 'paragraph',
      dimensions: 768,
    }},
  ],
  fields: [
    { name: 'title', type: { name: 'text' }, is_required: true },
    { name: 'body', type: { name: 'text' } },
  ],
}

Configuration Reference

Knative Worker Stack

The runtime consists of three packages:

Job Flow

1. Trigger fires -> inserts row into app_jobs.jobs
2. Worker polls -> picks up job by task_identifier
3. Worker POSTs -> ${KNATIVE_SERVICE_URL}/${task_identifier} with JSON payload
4. Function executes -> returns success/failure
5. Worker updates -> marks job as complete or failed (retries up to max_attempts)

Headers sent to the function:

• X-Worker-Id — worker instance identifier
• X-Job-Id — job row ID
• X-Database-Id — database context (nullable)
• X-Actor-Id — user who triggered the job (nullable)

Key Environment Variables

Scheduled Jobs

For recurring jobs, use app_jobs.add_scheduled_job() or the runtime_schedules table (in agentic-db):

-- database_id and actor_id are read from JWT claims automatically
SELECT app_jobs.add_scheduled_job(
  identifier := 'daily_report',
  payload := '{"report_type": "daily"}'::json,
  schedule_info := json_build_object(
    'rule', '0 9 * * *'  -- 9 AM daily
  )
);

The scheduler component in knative-job-service evaluates cron expressions and enqueues jobs at the appropriate times.

References

Cross-References

• Cloud functions (Knative handlers): constructive-platform
• Security policies: constructive-security
• AI and embeddings: constructive-agents
• Events (shared conditions system): constructive-events
• Blueprint definition format: constructive-blueprints