Skip to content

Background Jobs & Worker Database Access

This document explains how asynchronous background workers and scheduled tasks communicate with PostgreSQL.

The Core Problem

When a human user interacts with an app, their request carries a JWT token with their identity, which PostgREST and Row-Level Security (RLS) use to filter data.

However, background workers (OCR text extraction, document rendering, bank transaction syncs, agent workspace tasks) run independently. They carry no user identity and have no user token.

The Rule: Direct asyncpg & *_for_worker Functions

  1. Direct Connection, No HTTP Proxies: Workers connect directly to PostgreSQL via connection pools (asyncpg) using the service's constrained database login (e.g. internal_api_login). Workers never call PostgREST over HTTP and never mint fake client-credentials tokens to call their own database.
  2. Private Queue Functions (*_for_worker): Workers interact with queues and state machines through dedicated SQL functions in private domain schemas (e.g. invoicing.claim_receipt_extraction_jobs_for_worker).
  3. Lease-Token Fencing: Queue functions use optimistic concurrency locks (lease tokens) to prevent duplicate processing if a job is picked up by multiple workers simultaneously.

The *_for_worker SQL Contract

Every worker database function adheres to these standards:

  • Private Schema: Resides in a domain schema (invoicing, media, etc.) that is completely excluded from PostgREST (PGRST_DB_SCHEMAS). It is physically unreachable via HTTP.
  • Security Definer: Marked SECURITY DEFINER with SET search_path = pg_catalog to prevent search path hijacking.
  • Role Gate: Guarded with pg_has_role(session_user, 'internal_api_role', 'member') instead of reading JWT claims.
  • Explicit Revocation: Execution is revoked from PUBLIC and granted only to the specific service role.

Technical Pointers

  • Scheduled Jobs & Worker Loops: src/vibrant_platform_internal_api/jobs/
  • Receipt Processing: jobs/receipt_processing.py
  • Amex / Plaid Sync: jobs/amex_sync.py
  • Connecteam Time Sync: jobs/connecteam_sync.py
  • Worker Repositories: src/vibrant_platform_internal_api/modules/invoicing/
  • receipt_extraction_repository.py
  • document_rendition_repository.py
  • Agent Manager Workflows: src/vibrant_platform_agent_manager/
  • SQL Migration Reference: infrastructure/db/migrations/20260901000100_private_worker_rpcs.sql