n8nworkflows.xyz/workflows/Process OpenAI batch requests with a Supabase-Postgres FIFO queue-13238/readme-13238.md

Process OpenAI batch requests with a Supabase/Postgres FIFO queue

https://n8nworkflows.xyz/workflows/process-openai-batch-requests-with-a-supabase-postgres-fifo-queue-13238


# Process OpenAI batch requests with a Supabase/Postgres FIFO queue

## 1. Workflow Overview

**Purpose:**  
Process multiple OpenAI requests cheaply via the **OpenAI Batch API**, while tracking jobs in a persistent **Supabase/Postgres FIFO queue**. The workflow has two phases:

### 1.1 Phase 1 — Submit Batch (manual / upstream-triggered)
Takes a `systemPrompt` plus an array of `inputs`, converts them into OpenAI Batch `.jsonl` format, uploads the file to OpenAI **Files API**, creates a batch job via **Batches API**, then stores the batch `id` and initial status in Supabase (`openai_batches`).

### 1.2 Phase 2 — Poll & Retrieve (scheduled)
Every 5 minutes, pulls the earliest batch whose status is *not* terminal, retrieves the batch status from OpenAI, and:
- If **not completed**: updates status in Supabase.
- If **completed** (and has `output_file_id`): downloads output `.jsonl`, parses it, extracts responses, and updates Supabase with status + results.

---

## 2. Block-by-Block Analysis

### Block 1 — Submit Batch Input Preparation & Upload
**Overview:** Converts an array of prompts into the OpenAI Batch `.jsonl` format, then uploads that `.jsonl` to OpenAI Files API for batch processing.

**Nodes involved:**
- `Start (mock data)`
- `Convert to batch requests in .jsonl`
- `Call Files API`

#### Node: Start (mock data)
- **Type / role:** Manual Trigger (`manualTrigger`) to simulate upstream input.
- **Configuration choices:** Uses pinned data with:
  - `systemPrompt` (string)
  - `inputs` (array of strings)
- **Inputs/Outputs:** Entry point → outputs one item containing the mock JSON.
- **Edge cases:**
  - If `inputs` is missing or not an array, downstream Code node fails.
  - If prompts contain unexpected types (objects), `.jsonl` line creation may still serialize but the OpenAI API may reject content.

#### Node: Convert to batch requests in .jsonl
- **Type / role:** Code node (`code`) that generates OpenAI Batch `.jsonl` and returns it as **binary** (base64).
- **Key logic/config:**
  - Reads:
    - `const inputs = $input.first().json.inputs`
    - `const systemPrompt = $input.first().json.systemPrompt`
  - Builds one JSON line per input with:
    - `custom_id: `${index}_${Date.now()}``
    - `method: "POST"`
    - `url: "/v1/responses"`
    - `body: { model: "gpt-5-mini", instructions: systemPrompt, input }`
  - Joins lines with `\n` to create NDJSON.
  - Base64-encodes content and returns as:
    - `binary.data.data` (base64)
    - `mimeType: application/x-ndjson`
    - `fileName: batch_input.jsonl`
- **Connections:** Input from `Start (mock data)` → output to `Call Files API`.
- **Version notes:** Code node v2; uses Node.js `Buffer` (available in n8n Code node).
- **Edge cases / failure modes:**
  - Large `inputs` can create oversized `.jsonl` → OpenAI file upload limits or timeouts.
  - `Date.now()` inside map can produce identical timestamps within same ms; uniqueness relies also on `index`.
  - Uses `/v1/responses` format; if your OpenAI account/model doesn’t support Responses API in batch, batch creation will fail.

#### Node: Call Files API
- **Type / role:** HTTP Request node uploading the `.jsonl` to OpenAI Files.
- **Configuration choices:**
  - `POST https://api.openai.com/v1/files`
  - `multipart-form-data`
  - Form fields:
    - `purpose = batch`
    - `file` = binary field `data`
  - Auth: `predefinedCredentialType` → `openAiApi` credential
  - `alwaysOutputData: true` (so it still outputs something on some error conditions)
- **Inputs/Outputs:**
  - Expects binary `data` from previous node.
  - Outputs OpenAI file object (notably `id`) used by the next step.
- **Edge cases / failure modes:**
  - Missing/invalid OpenAI credential → 401/403.
  - Incorrect binary field name (must be `data`) → empty upload or 400.
  - Network timeout / file too large → request fails.

---

### Block 2 — Create Batch Job & Persist in Supabase
**Overview:** Creates the OpenAI batch using the uploaded file id, then inserts a tracking row in Supabase.

**Nodes involved:**
- `Call Batch API`
- `Create a row in batch table`
- `Submission done`

#### Node: Call Batch API
- **Type / role:** HTTP Request node to create a batch job.
- **Configuration choices:**
  - `POST https://api.openai.com/v1/batches`
  - JSON body:
    - `input_file_id: {{ $json.id }}` (from Files API response)
    - `endpoint: "/v1/responses"`
    - `completion_window: "24h"`
  - Auth: OpenAI credential
  - `alwaysOutputData: true`
- **Connections:** From `Call Files API` → to `Create a row in batch table`.
- **Edge cases / failure modes:**
  - If `$json.id` is missing (failed upload), body becomes invalid.
  - Endpoint mismatch: file lines target `/v1/responses` and batch endpoint also `/v1/responses`—if you change one, you must change the other.
  - OpenAI may reject unsupported models or invalid request schema in the `.jsonl`.

#### Node: Create a row in batch table
- **Type / role:** Supabase node (`supabase`) to insert (or create) tracking record.
- **Configuration choices:**
  - Table: `openai_batches`
  - Fields set:
    - `id = {{ $json.id }}`
    - `batch_status = {{ $json.status }}`
- **Connections:** From `Call Batch API` → to `Submission done`.
- **Credentials:** `supabaseApi` (named `b2b`)
- **Edge cases / failure modes:**
  - If `id` is PK and already exists → insert conflict (depends on Supabase node behavior; likely fails unless configured for upsert).
  - Schema mismatch: if `openai_batches` table/columns differ, update fails.
  - `batch_status` null not allowed per note; if OpenAI response lacks `status`, insert fails.

#### Node: Submission done
- **Type / role:** NoOp node marking end of submit phase.
- **Connections:** From `Create a row in batch table`.
- **Edge cases:** None (pure terminator/visual marker).

---

### Block 3 — FIFO Polling: Select Next Batch to Check
**Overview:** On a schedule, selects batches that are not terminal (completed/failed/expired/cancelled), ordered by creation time (FIFO intent), then requests status from OpenAI.

**Nodes involved:**
- `Cron Job (5 mins)`
- `Get the earliest uncompleted batch`
- `Call Batch API to retrieve batch object`

#### Node: Cron Job (5 mins)
- **Type / role:** Schedule Trigger (`scheduleTrigger`) entry point for polling.
- **Configuration choices:** Interval: every minute field (configured as minutes interval; effectively “every 5 mins” per node name, but JSON shows interval with `{field:"minutes"}` and no explicit “5” value).
- **Connections:** → `Get the earliest uncompleted batch`
- **Edge cases:**
  - If interval is misconfigured (e.g., every 1 minute), may spam OpenAI status checks.
  - If too infrequent, completed batches will sit longer before retrieval.

#### Node: Get the earliest uncompleted batch
- **Type / role:** Postgres node (`postgres`) selecting candidate batch rows.
- **Configuration choices:**
  - Schema: `public`
  - Table: `openai_batches`
  - Sort: by `created_at` ascending (FIFO ordering)
  - Where filters (all applied):
    - `batch_status != failed`
    - `batch_status != expired`
    - `batch_status != cancelled`
    - `batch_status != completed`
  - `returnAll: true` (returns all matching rows, not just the earliest)
- **Connections:** → `Call Batch API to retrieve batch object`
- **FIFO note:** Despite node name, **it does not `LIMIT 1`**, so multiple batches may be processed in a single run (n8n will execute downstream once per item).
- **Edge cases / failure modes:**
  - If table is empty → outputs no items; downstream doesn’t run (expected).
  - If statuses include other terminal values not excluded, they’ll keep being polled.
  - If many rows returned, you may hit OpenAI rate limits with many status calls per trigger.
  - Credential/network errors to Postgres.

#### Node: Call Batch API to retrieve batch object
- **Type / role:** HTTP Request to fetch batch status/details from OpenAI.
- **Configuration choices:**
  - `GET https://api.openai.com/v1/batches/{{ $json.id }}`
  - Auth: OpenAI credential
- **Connections:** → `If status = completed`
- **Edge cases / failure modes:**
  - If `$json.id` missing/blank → invalid URL.
  - 404 if batch id doesn’t exist (row corruption/manual edits).
  - 401/403 auth issues.

---

### Block 4 — Branch: Completed vs Not Completed
**Overview:** If batch is completed and has an output file, download and parse results; otherwise update only the batch status.

**Nodes involved:**
- `If status = completed`
- `Update status` (not completed path)
- `Download .jsonl result` (completed path)

#### Node: If status = completed
- **Type / role:** IF node branching on OpenAI batch object fields.
- **Conditions (AND):**
  1. `{{ $json.status }}` equals `completed`
  2. `{{ $json.output_file_id }}` exists
- **Outputs:**
  - **True** → `Download .jsonl result`
  - **False** → `Update status`
- **Edge cases:**
  - Some batches could be `completed` but have no `output_file_id` due to API changes/errors → goes to “false” branch and only updates status.
  - If OpenAI returns status `failed/expired/cancelled`, it will go “false” and be updated, but will also be excluded from future polling by the SQL filter.

#### Node: Update status
- **Type / role:** Supabase update to persist latest status for non-completed (or completed-without-output) batches.
- **Configuration choices:**
  - Table: `openai_batches`
  - Filter: `id eq {{ $json.id }}`
  - Updates:
    - `updated_at = {{ $now }}`
    - `batch_status = {{ $json.status }}`
- **Connections:** None further (ends that branch).
- **Edge cases / failure modes:**
  - If row does not exist → update affects 0 rows (silent or error depending on node behavior).
  - `$now` formatting: Supabase expects timestamp; n8n typically provides ISO string—ensure column type is compatible (`timestamptz` is fine).

#### Node: Download .jsonl result
- **Type / role:** HTTP Request to download the output file content from OpenAI Files API.
- **Configuration choices:**
  - `GET https://api.openai.com/v1/files/{{ $json.output_file_id }}/content`
  - Response format: `file` (binary)
  - Auth: OpenAI credential
- **Connections:** → `.jsonl to base64`
- **Edge cases / failure modes:**
  - Output file not ready immediately (rare if status says completed, but possible timing issues) → 404/409.
  - Large output file → memory pressure in n8n when handled as binary.

---

### Block 5 — Decode Output, Extract Results, Persist
**Overview:** Converts the downloaded NDJSON file into text, parses each line, extracts the assistant message text from each response, then writes results back into Supabase.

**Nodes involved:**
- `.jsonl to base64`
- `Decode base64`
- `Update status and result`
- `Retrieval done`

#### Node: .jsonl to base64
- **Type / role:** Extract From File node (`extractFromFile`) converting binary content into a JSON property.
- **Configuration choices:**
  - Operation: `binaryToPropery` (note the node’s operation name; intended meaning is “binary to property”)
- **Inputs/Outputs:**
  - Input: binary file from `Download .jsonl result`
  - Output: JSON field `data` holding base64 (as used by next node)
- **Edge cases / failure modes:**
  - If binary property name differs from default (usually `data`), conversion may fail or produce empty output.
  - Very large files increase memory usage.

#### Node: Decode base64
- **Type / role:** Code node parsing the NDJSON batch output and extracting texts.
- **Key logic/config:**
  - Reads base64 from `const base64Content = $input.first().json.data;`
  - Decodes to UTF-8, splits by newline, `JSON.parse` per line (skipping invalid lines)
  - For each parsed line:
    - Skips if `line.error` exists or `line.response.status_code !== 200`
    - Extracts `line.response.body.output`
    - Finds element where `type === 'message'`
    - Extracts `messageOutput.content[0].text`
  - Counts `parsed` and `errors`
  - Pulls batch metadata from earlier node:
    - `const batch = $('If status = completed').first().json;`
  - Outputs:
    - `id`, `status`, `output_file_id`, `result` (array of texts), `parsed`, `errors`
- **Connections:** → `Update status and result`
- **Version notes:** Code node v2; uses `Buffer` and the `$('Node Name')` selector.
- **Edge cases / failure modes:**
  - If OpenAI output schema differs (e.g., content is not `[ {text: ...} ]`) → extraction yields empty results and increments errors.
  - If multiple message parts exist, only the first `content[0].text` is taken.
  - If `$('If status = completed')` has no item (unexpected execution path), `.first()` will throw.
  - NDJSON lines that fail JSON parsing are dropped.

#### Node: Update status and result
- **Type / role:** Supabase update to store final status and extracted results.
- **Configuration choices:**
  - Filter: `id eq {{ $json.id }}`
  - Updates:
    - `updated_at = {{ $now }}`
    - `batch_status = {{ $json.status }}`
    - `output_file_id = {{ $json.output_file_id }}`
    - `result = {{ $json.result }}`
- **Connections:** → `Retrieval done`
- **Edge cases / failure modes:**
  - `result` column type must accept an array (per sticky note: `text[]`). If set as `text`/JSON incorrectly, update fails.
  - Large arrays may hit row size or request limits.

#### Node: Retrieval done
- **Type / role:** NoOp node marking end of retrieval phase.
- **Connections:** From `Update status and result`.

---

## 3. Summary Table

| Node Name | Node Type | Functional Role | Input Node(s) | Output Node(s) | Sticky Note |
|---|---|---|---|---|---|
| Start (mock data) | Manual Trigger | Entry point for submit phase with sample inputs | — | Convert to batch requests in .jsonl | ## 📤 Phase 1: Submit Batch |
| Convert to batch requests in .jsonl | Code | Build NDJSON batch file and output as binary | Start (mock data) | Call Files API | ## 📤 Phase 1: Submit Batch |
| Call Files API | HTTP Request | Upload `.jsonl` file to OpenAI Files API | Convert to batch requests in .jsonl | Call Batch API | ## 📤 Phase 1: Submit Batch |
| Call Batch API | HTTP Request | Create OpenAI batch job | Call Files API | Create a row in batch table | ## 📤 Phase 1: Submit Batch |
| Create a row in batch table | Supabase | Insert batch record into `openai_batches` | Call Batch API | Submission done | ## 📤 Phase 1: Submit Batch |
| Submission done | NoOp | End marker for submission | Create a row in batch table | — | ## 📤 Phase 1: Submit Batch |
| Cron Job (5 mins) | Schedule Trigger | Entry point for polling phase | — | Get the earliest uncompleted batch | ## 📥 Phase 2: Poll & Retrieve (cron) |
| Get the earliest uncompleted batch | Postgres | Select not-terminal batches ordered by created time | Cron Job (5 mins) | Call Batch API to retrieve batch object | ## 📥 Phase 2: Poll & Retrieve (cron) |
| Call Batch API to retrieve batch object | HTTP Request | Fetch batch status/details from OpenAI | Get the earliest uncompleted batch | If status = completed | ## 📥 Phase 2: Poll & Retrieve (cron) |
| If status = completed | IF | Branch completed vs not completed | Call Batch API to retrieve batch object | Download .jsonl result (true), Update status (false) | ## 📥 Phase 2: Poll & Retrieve (cron) |
| Update status | Supabase | Update status for non-completed batch | If status = completed (false) | — | ## 📥 Phase 2: Poll & Retrieve (cron) |
| Download .jsonl result | HTTP Request | Download output file content (binary) | If status = completed (true) | .jsonl to base64 | ## 📥 Phase 2: Poll & Retrieve (cron) |
| .jsonl to base64 | Extract From File | Convert downloaded binary to base64 in JSON | Download .jsonl result | Decode base64 | ## 📥 Phase 2: Poll & Retrieve (cron) |
| Decode base64 | Code | Parse NDJSON and extract response texts | .jsonl to base64 | Update status and result | ## 📥 Phase 2: Poll & Retrieve (cron) |
| Update status and result | Supabase | Persist final status + output file id + results array | Decode base64 | Retrieval done | ## 📥 Phase 2: Poll & Retrieve (cron) |
| Retrieval done | NoOp | End marker for retrieval | Update status and result | — | ## 📥 Phase 2: Poll & Retrieve (cron) |

**Global sticky note content applicable to the whole workflow (context):**  
“OpenAI Batch API (FIFO with Supabase / Postgres)” note describing table schema, credentials, customization, and FIFO `LIMIT 1` recommendation.

---

## 4. Reproducing the Workflow from Scratch

1. **Create database table in Supabase (and ensure Postgres access):**
   - Table name: `openai_batches`
   - Columns:
     - `id` (text, primary key)
     - `batch_status` (text, not null)
     - `output_file_id` (text, nullable)
     - `result` (text[], nullable)
     - `created_at` (timestamptz, default `now()`)
     - `updated_at` (timestamptz, nullable)

2. **Create/Open credentials in n8n:**
   - **OpenAI API credential** (used by all HTTP Request nodes to `api.openai.com`)
   - **Supabase API credential** (used by Supabase nodes)
   - **Postgres credential** pointing to the Supabase Postgres database (used by Postgres node)

3. **Phase 1 nodes (Submit):**
   1. Add **Manual Trigger** named `Start (mock data)`.
      - Provide test pinned data (optional) with:
        - `systemPrompt` (string)
        - `inputs` (array of strings)
   2. Add **Code** node named `Convert to batch requests in .jsonl`.
      - Paste logic to:
        - Convert `inputs` + `systemPrompt` into NDJSON lines targeting `/v1/responses`
        - Output as `binary.data` with base64 content and filename `batch_input.jsonl`
      - Set model (e.g., `gpt-5-mini`) in the code.
   3. Add **HTTP Request** node named `Call Files API`.
      - Method: `POST`
      - URL: `https://api.openai.com/v1/files`
      - Authentication: **OpenAI predefined credential**
      - Body: `multipart/form-data`
        - Field `purpose` = `batch`
        - Field `file` = **binary** from input, property name `data`
   4. Add **HTTP Request** node named `Call Batch API`.
      - Method: `POST`
      - URL: `https://api.openai.com/v1/batches`
      - Authentication: OpenAI credential
      - Body: JSON:
        - `input_file_id` = expression `{{ $json.id }}`
        - `endpoint` = `"/v1/responses"`
        - `completion_window` = `"24h"`
   5. Add **Supabase** node named `Create a row in batch table`.
      - Operation: Insert/Create row
      - Table: `openai_batches`
      - Fields:
        - `id` = `{{ $json.id }}`
        - `batch_status` = `{{ $json.status }}`
   6. Add **NoOp** node `Submission done`.

   7. **Connect Phase 1:**  
      `Start (mock data)` → `Convert to batch requests in .jsonl` → `Call Files API` → `Call Batch API` → `Create a row in batch table` → `Submission done`

4. **Phase 2 nodes (Poll & Retrieve):**
   1. Add **Schedule Trigger** named `Cron Job (5 mins)`.
      - Configure interval to **every 5 minutes** (ensure the value is actually 5 in the UI).
   2. Add **Postgres** node named `Get the earliest uncompleted batch`.
      - Operation: Select from `public.openai_batches`
      - Filter: exclude terminal statuses:
        - `batch_status != completed`
        - `batch_status != failed`
        - `batch_status != expired`
        - `batch_status != cancelled`
      - Sort by `created_at` ascending
      - For strict FIFO, set **Limit = 1** (recommended; otherwise it will process all matching rows).
   3. Add **HTTP Request** node named `Call Batch API to retrieve batch object`.
      - Method: `GET`
      - URL: `https://api.openai.com/v1/batches/{{ $json.id }}`
      - Authentication: OpenAI credential
   4. Add **IF** node named `If status = completed`.
      - Condition 1: `{{ $json.status }}` equals `completed`
      - Condition 2: `{{ $json.output_file_id }}` exists
      - Combine with AND
   5. False branch (not completed): add **Supabase** node `Update status`
      - Operation: Update
      - Table: `openai_batches`
      - Filter: `id eq {{ $json.id }}`
      - Fields:
        - `updated_at = {{ $now }}`
        - `batch_status = {{ $json.status }}`
   6. True branch (completed): add **HTTP Request** node `Download .jsonl result`
      - Method: `GET`
      - URL: `https://api.openai.com/v1/files/{{ $json.output_file_id }}/content`
      - Authentication: OpenAI credential
      - Response: **File** (binary)
   7. Add **Extract From File** node named `.jsonl to base64`
      - Operation: “binary to property”
      - Keep default binary property name `data` unless you changed it.
   8. Add **Code** node `Decode base64`
      - Decode base64 to string
      - Parse NDJSON lines
      - Extract `message` output text from each line
      - Also reference the completed batch object via `$('If status = completed').first().json`
   9. Add **Supabase** node `Update status and result`
      - Operation: Update
      - Table: `openai_batches`
      - Filter: `id eq {{ $json.id }}`
      - Fields:
        - `updated_at = {{ $now }}`
        - `batch_status = {{ $json.status }}`
        - `output_file_id = {{ $json.output_file_id }}`
        - `result = {{ $json.result }}`
   10. Add **NoOp** node `Retrieval done`.

   11. **Connect Phase 2:**  
      `Cron Job (5 mins)` → `Get the earliest uncompleted batch` → `Call Batch API to retrieve batch object` → `If status = completed`  
      - True: `Download .jsonl result` → `.jsonl to base64` → `Decode base64` → `Update status and result` → `Retrieval done`  
      - False: `Update status`

---

## 5. General Notes & Resources

| Note Content | Context or Link |
|---|---|
| OpenAI Batch API (FIFO with Supabase / Postgres): submit `.jsonl` file, create batch, store status in `openai_batches`, cron polls earliest uncompleted, downloads output and stores results. | Workflow design note (global) |
| Supabase table schema required: `id` (text PK), `batch_status` (text), `output_file_id` (text nullable), `result` (text[] nullable), `created_at` default now(), `updated_at` nullable. | Database prerequisite |
| Customization: change model in “Convert to batch requests…” (currently `gpt-5-mini`), adjust cron interval, and for strict FIFO set Postgres select to `LIMIT 1`. | Operational tuning |

disclaimer Le texte fourni provient exclusivement d’un workflow automatisé réalisé avec n8n, un outil d’intégration et d’automatisation. Ce traitement respecte strictement les politiques de contenu en vigueur et ne contient aucun élément illégal, offensant ou protégé. Toutes les données manipulées sont légales et publiques.