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`

Common Job Trigger Patterns

Full blueprint examples for common job trigger scenarios.

1. Email on Invite Creation

Send an email when a new invite is inserted:

{
  ref: 'invites',
  table_name: 'invites',
  nodes: [
    ...ORG_NODES,
    {
      $type: 'JobTrigger',
      data: {
        task_identifier: 'send_invite_email',
        payload_strategy: 'fields',
        payload_fields: ['id', 'email', 'role'],
        events: ['INSERT'],
      },
    },
  ],
  fields: [
    { name: 'email', type: { name: 'citext' }, is_required: true },
    { name: 'role', type: { name: 'text' }, default_value: { value: 'member' } },
    { name: 'accepted_at', type: { name: 'timestamptz' } },
  ],
}

The send_invite_email Knative function receives:

{ "id": "abc-123", "email": "user@example.com", "role": "member" }

2. External System Sync (Stripe)

Sync data whenever specific fields change:

{
  ref: 'invoices',
  table_name: 'invoices',
  nodes: [
    ...ORG_NODES,
    {
      $type: 'JobTrigger',
      data: {
        task_identifier: 'sync_to_stripe',
        payload_strategy: 'fields',
        payload_fields: ['id', 'amount', 'currency', 'status'],
        events: ['INSERT', 'UPDATE'],
        watch_fields: ['amount', 'status'],
        queue_name: 'stripe',
        max_attempts: 5,
      },
    },
  ],
  fields: [
    { name: 'amount', type: { name: 'numeric' }, is_required: true },
    { name: 'currency', type: { name: 'text' }, default_value: { value: 'USD' } },
    { name: 'status', type: { name: 'text' }, default_value: { value: 'draft' } },
    { name: 'stripe_id', type: { name: 'text' } },
  ],
}

watch_fields means the UPDATE trigger only fires when amount or status actually change.

3. Conditional Trigger (Fire on Status Value)

Only fire when a specific field has a specific value:

{
  $type: 'JobTrigger',
  data: {
    task_identifier: 'publish_to_cdn',
    events: ['UPDATE'],
    condition_field: 'status',
    condition_value: 'published',
  },
}

Creates a WHEN clause: WHEN (NEW.status = 'published'). The trigger only fires on UPDATE when status equals 'published'.

Note: condition_field and watch_fields cannot both be specified.

3b. Compound Conditions (Status Transition)

Fire only when a row transitions from one status to another:

{
  $type: 'JobTrigger',
  data: {
    task_identifier: 'process_published',
    events: ['UPDATE'],
    payload_strategy: 'custom',
    payload_custom: { doc_id: 'id', title: 'title' },
    conditions: [
      { field: 'status', op: '=', value: 'published' },
      { field: 'status', op: '=', value: 'draft', row: 'OLD' },
    ],
  },
}

Creates a WHEN clause: WHEN (NEW.status = 'published' AND OLD.status = 'draft'). The trigger only fires when status changes from 'draft' to 'published'.

3c. Compound Conditions with OR (MIME Type Filtering)

Fire when status transitions AND the row matches one of several MIME patterns:

{
  $type: 'JobTrigger',
  data: {
    task_identifier: 'process_media',
    events: ['UPDATE'],
    payload_strategy: 'custom',
    payload_custom: { file_id: 'id', key: 'key', mime_type: 'mime_type' },
    include_meta: true,
    conditions: {
      AND: [
        { field: 'status', op: '=', value: 'ready' },
        { field: 'status', op: '=', value: 'pending', row: 'OLD' },
        { OR: [
          { field: 'mime_type', op: 'LIKE', value: 'image/%' },
          { field: 'mime_type', op: 'LIKE', value: 'video/%' },
        ]},
      ]
    },
  },
}

3d. ProcessImageEmbedding (Composition Shorthand)

For the common pattern of embedding image files on insert, use ProcessImageEmbedding instead of manually wiring SearchVector + JobTrigger:

nodes: [
  ...STORAGE_NODES,
  { $type: 'ProcessImageEmbedding' },
]

Equivalent to manually configuring SearchVector (512-dim, HNSW, cosine) + JobTrigger (INSERT, mime_type LIKE 'image/%'). Override defaults as needed:

{
  $type: 'ProcessImageEmbedding',
  data: {
    dimensions: 1024,
    metric: 'l2',
    mime_patterns: ['image/%', 'video/%'],
    task_identifier: 'custom_embedding_worker',
  },
}

4. Audit Trail on Delete

Capture full row data before deletion:

{
  $type: 'JobTrigger',
  data: {
    task_identifier: 'audit_document_delete',
    payload_strategy: 'row',
    events: ['DELETE'],
    include_meta: true,
  },
}

5. Debounced Batch Processing

Use job_key + run_at_delay to debounce rapid changes into a single job:

{
  $type: 'JobTrigger',
  data: {
    task_identifier: 'aggregate_analytics',
    events: ['INSERT'],
    job_key: 'aggregate_analytics_batch',
    run_at_delay: '5 minutes',
    queue_name: 'analytics',
    priority: 10,
  },
}

job_key gives the job upsert semantics — subsequent inserts reset the run_at timer instead of creating duplicate jobs.

6. Webhook Dispatch with Custom Payload

Reshape column names for an external webhook:

{
  $type: 'JobTrigger',
  data: {
    task_identifier: 'dispatch_webhook',
    payload_strategy: 'custom',
    payload_custom: {
      order_id: 'id',
      total_amount: 'total',
      customer: 'customer_id',
      event_type: 'status',
    },
    events: ['INSERT', 'UPDATE'],
    watch_fields: ['status', 'total'],
  },
}

7. Multiple Triggers on One Table

A single table can have several JobTrigger nodes for independent workflows:

nodes: [
  ...ORG_NODES,
  {
    $type: 'JobTrigger',
    data: {
      task_identifier: 'sync_to_hubspot',
      events: ['INSERT', 'UPDATE'],
      watch_fields: ['email', 'first_name', 'last_name'],
      queue_name: 'crm_sync',
    },
  },
  {
    $type: 'JobTrigger',
    data: {
      task_identifier: 'send_welcome_email',
      events: ['INSERT'],
    },
  },
  {
    $type: 'JobTrigger',
    data: {
      task_identifier: 'audit_contact_delete',
      payload_strategy: 'row',
      events: ['DELETE'],
    },
  },
],

8. Embedding Generation Note

SearchVector and SearchUnified nodes already auto-create embedding job triggers when enqueue_job: true (the default). Use JobTrigger only for custom processing beyond embedding generation:

nodes: [
  ...ORG_NODES,
  { $type: 'SearchUnified', data: {
    embedding: { source_fields: ['title', 'content'] },
    bm25: { field_name: 'embedding_text' },
  }},
  // Separate trigger for a different pipeline
  { $type: 'JobTrigger', data: {
    task_identifier: 'classify_document',
    events: ['INSERT'],
    payload_strategy: 'fields',
    payload_fields: ['id', 'title', 'content'],
  }},
],

Payload Strategies

Detailed examples of each JobTrigger payload strategy.

row_id (default) — Just the Row ID

The lightest payload. The function fetches full data via GraphQL as needed.

{
  $type: 'JobTrigger',
  data: { task_identifier: 'process_invoice' }
}
// payload: { "id": "abc-123" }

Best for: most use cases where the function needs fresh data anyway.

row — Full Row as JSON

Sends the entire NEW (or OLD for DELETE) row.

{
  $type: 'JobTrigger',
  data: {
    task_identifier: 'audit_change',
    payload_strategy: 'row',
    events: ['INSERT', 'UPDATE', 'DELETE'],
  }
}
// INSERT payload: { "id": "...", "amount": 100, "status": "paid", "created_at": "..." }
// DELETE payload: { "id": "...", "amount": 100, "status": "paid", ... } (OLD row)

Add include_old: true to also get the previous row on UPDATE:

{
  $type: 'JobTrigger',
  data: {
    task_identifier: 'diff_audit',
    payload_strategy: 'row',
    events: ['UPDATE'],
    include_old: true,
  }
}
// payload: { "new": { ... }, "old": { ... } }

Best for: audit trails, full-context processing, diffing old vs new values.

fields — Selected Columns Only

Sends only the columns you specify. Reduces payload size.

{
  $type: 'JobTrigger',
  data: {
    task_identifier: 'sync_to_stripe',
    payload_strategy: 'fields',
    payload_fields: ['id', 'amount', 'currency', 'status'],
    events: ['INSERT'],
  }
}
// payload: { "id": "...", "amount": 100, "currency": "USD", "status": "draft" }

Best for: external API sync where only specific fields matter.

custom — Mapped Key Names

Renames columns in the payload. Useful when the external system expects different field names.

{
  $type: 'JobTrigger',
  data: {
    task_identifier: 'webhook_fire',
    payload_strategy: 'custom',
    payload_custom: {
      invoice_id: 'id',
      total: 'amount',
      state: 'status',
    },
    events: ['INSERT'],
  }
}
// payload: { "invoice_id": "...", "total": 100, "state": "draft" }

Best for: webhook dispatch, external API integration with specific payload shapes.

Adding Metadata

Any strategy can include table/schema metadata with include_meta: true:

{
  $type: 'JobTrigger',
  data: {
    task_identifier: 'generic_audit',
    payload_strategy: 'row_id',
    include_meta: true,
    events: ['INSERT', 'UPDATE', 'DELETE'],
  }
}
// payload: { "id": "...", "_meta": { "schema": "app_public", "table": "invoices", "event": "INSERT" } }

Best for: generic handlers that process events from multiple tables.