Datasets

Dataset-level task handlers — generating, importing, updating, deleting, duplicating, and cleaning up datasets and their attachments.

Overview

The worker supports seven dataset-level task types. Unlike ai_generate_records and ai_generate_fields, which operate on the contents and schema of an existing dataset, these tasks create, modify, or manage datasets themselves.

Database Tables

datasets Table

The primary table for dataset metadata. Dataset-level handlers read from and write to this table. The fields column holds the complete field schema as a JSONB array.

dataset_shares Table

Controls who has access to a dataset. When any handler creates a new dataset, it must immediately insert an owner share for task.created_by or the user will not be able to see it.

pending_attachments Table

Tracks file uploads that are waiting to be associated with a record. Rows are written by the API server when a user uploads a file but hasn't yet saved the record. The cleanup_stale_attachments handler removes rows (and their storage files) that have been waiting too long.

ai_generate_datasets

Generates new dataset schemas via AI (generate mode) or updates the metadata of existing datasets (enrich mode). Both modes use a structured prompt and return AI-generated field definitions post-processed by restructureAIFields — assigning fld_ IDs and nesting options into a sub-object.

Input Schema

{
  "mode": "generate",
  "count": 3,
  "ai": {
    "prompt": "Generate SaaS product datasets with pricing and feature data",
    "model": "gpt-4o",
    "temperature": 0.7,
    "max_output_tokens": 16000
  }
}

{
  "mode": "enrich",
  "datasetIds": ["ds_abc123", "ds_def456"],
  "ai": {
    "prompt": "Add tags and improve descriptions based on the existing fields",
    "model": "gpt-4o",
    "temperature": 0.5
  }
}

Generate Mode

A single AI call produces all count datasets at once using structured output (GENERATE_BASED_SCHEMA). Each dataset is then inserted into datasets and an owner share is created in dataset_shares. If an individual insert fails, that dataset is recorded with an error but the rest continue (partial success within generate mode).

Enrich Mode

Each dataset in datasetIds is enriched with a separate AI call (ENRICH_DATASET_SCHEMA). The current dataset state (name, description, tags, field names, and up to 5 sample records) is passed to the model. Only non-null fields returned by the model are applied. Cancellation is checked between datasets.

Output Schema

{
  "datasetIds": ["ds_xyz001", "ds_xyz002"],
  "datasets": [
    {
      "id": "ds_xyz001",
      "name": "SaaS Pricing Data",
      "description": "Pricing plans for SaaS companies",
      "tags": ["saas", "pricing"],
      "fields": [...],
      "tokenUsage": { "input_tokens": 1200, "output_tokens": 900, "total_tokens": 2100 }
    }
  ],
  "tokenUsage": { "input_tokens": 1200, "output_tokens": 900, "total_tokens": 2100 }
}

import_dataset

Downloads a file from object storage, parses it, creates a new dataset, and batch-inserts the rows as records. The file is always deleted from storage at the end, whether the import succeeds or fails.

Input Schema

{
  "fileKey": "imports/user_abc/companies.csv",
  "name": "My Companies",
  "description": "Imported from CSV",
  "tags": ["import", "companies"],
  "options": {},
  "fields": [
    { "id": "fld_a1b2c3d4e5f6", "name": "Company", "type": "text", "options": { "required": true } },
    { "id": "fld_b2c3d4e5f6g7", "name": "Revenue", "type": "number" }
  ]
}

Supported Formats

Format is detected from the file extension:

Type Inference

When fields is not provided, the worker samples all values in each column and infers the most specific matching type. The inference priority is: boolean → integer → number → email → url → date → array → object → text. Inferred attachment type results are filtered out defensively.

Type Coercion

During validation, values from CSV files (which are always strings) are coerced to their field type. Numbers strip currency symbols. Booleans accept "true"/"false"/"1"/"0"/"yes"/"no". Dates accept ISO 8601, common locale formats, and Unix timestamps. Values that cannot be coerced are set to null and recorded as row-level errors.

Partial Success

Row-level validation errors do not abort the import. All valid rows are inserted; invalid rows are collected in output.errors. The task completes successfully as long as at least one row was inserted. If zero rows are valid, the task fails.

Output Schema

{
  "dataset": {
    "id": "ds_abc123",
    "name": "My Companies",
    "description": "Imported from CSV",
    "tags": ["import", "companies"],
    "fields": [...]
  },
  "detectedFormat": "csv",
  "fieldsSource": "inferred",
  "fields": [...],
  "summary": {
    "totalRows": 850,
    "imported": 847,
    "failed": 3
  },
  "errors": [
    { "index": 5, "error": "row 5: Revenue — expected number, got 'N/A'" },
    { "index": 23, "error": "row 23: Email — expected valid email address, got 'not-an-email'" }
  ]
}

batch_update_datasets

Updates name, description, tags, and/or options on multiple datasets sequentially. The options field is merged with the existing value (shallow merge) rather than replaced wholesale. Cancellation is checked between items.

Input Schema

{
  "datasetUpdates": [
    {
      "id": "ds_abc123",
      "name": "Renamed Dataset",
      "description": "Updated description",
      "tags": ["updated", "q1"],
      "options": { "locked": true }
    },
    {
      "id": "ds_def456",
      "tags": ["archived"]
    }
  ]
}

Partial Success

If individual updates fail (e.g., dataset not found), they are recorded in output.errors and the rest continue. The task completes successfully if at least one update succeeded. If all updates fail, the task is marked as failed.

Output Schema

{
  "updatedIds": ["ds_abc123"],
  "updated": 1,
  "errors": [
    { "id": "ds_def456", "error": "Dataset \"ds_def456\" not found" }
  ]
}

batch_delete_datasets

Deletes multiple datasets either as a soft-delete (sets deleted_at) or a permanent delete (hard DELETE that cascades to records, shares, and invitations). Cancellation is checked between items.

Input Schema

{
  "datasetIds": ["ds_abc123", "ds_def456", "ds_ghi789"],
  "permanent": false
}

Partial Success

Datasets that are not found or fail to delete are recorded in output.errors. The task completes if at least one deletion succeeded; fails if all deletions failed.

Output Schema

{
  "deletedIds": ["ds_abc123", "ds_def456"],
  "deleted": 2,
  "errors": [
    { "id": "ds_ghi789", "error": "Dataset \"ds_ghi789\" not found" }
  ]
}

duplicate_dataset

Creates a copy of a dataset with a new name. The new dataset is owned by task.created_by. Metadata is copied from the source but public and locked are always forced to false on the copy. Optionally copies all non-deleted records in pages of 500, with cancellation checks between pages.

Input Schema

{
  "sourceDatasetId": "ds_abc123",
  "name": "Copy of My Dataset",
  "description": "Optional override description",
  "include_records": true
}

Record Paging

Records are copied in pages of 500 rows ordered by created_at ASC. After each page, the task emits a records_copied event and updates progress (10–99%). Cancellation is checked at the start of each page. If cancelled mid-copy, the partially-copied records remain in the new dataset.

Output Schema

{
  "datasetId": "ds_xyz999",
  "name": "Copy of My Dataset",
  "recordsCopied": 1247
}

cleanup_stale_attachments

Finds rows in pending_attachments older than a configurable TTL, deletes the associated file from object storage, then deletes the database row. This task is designed to run on a schedule (e.g., every hour) to prevent orphaned uploads from accumulating.

Input Schema

{
  "ttlHours": 24
}

Cancellation

Cancellation is checked between each row. If cancelled mid-run, already-deleted rows remain deleted.

Output Schema

{
  "deleted": 12
}

cleanup_dataset_attachments

Lists and deletes all object storage files under the prefix attachments/{datasetId}/. This task is typically queued after a dataset is permanently deleted, to clean up its attachment files from storage. It does not query the database — the dataset rows are already gone by the time it runs.

Input Schema

{
  "datasetId": "ds_abc123"
}

Output Schema

{
  "deleted": 38
}