refactor: streamline repair dataset handling and enhance CLI functionality

- Updated the repair CLI to target a single IPFS-enabled dataset, simplifying dataset management.
- Removed obsolete dataset grouping logic and refactored related database queries for improved clarity.
- Enhanced error handling for missing datasets and improved command responses to include relevant dataset information.
- Introduced new utility functions for dataset retrieval and updated existing functions to align with the new dataset structure.
- Updated TypeScript types for better type safety and clarity in the codebase.

Author: hugomrdias

packages/repair-cli/AGENTS.md

@@ -13,45 +13,37 @@ Two databases:
 
 Commands get both via `contextMiddleware` (`middleware.ts`): wallet client from config, indexer URL by `chainId` (314 = mainnet, else calibration).
 
-## Piece groups
+## Repair dataset
 
-Pieces are grouped by dataset flags (`withCdn`, `withIpfsIndexing`). Groups are **mutually exclusive**:
+Repairs now target a single dataset: IPFS indexing enabled and CDN disabled.
 
-- `both`: `withCdn = true`, `withIpfsIndexing = true`
-- `cdn`: `withCdn = true`, `withIpfsIndexing = false`
-- `ipfs`: `withCdn = false`, `withIpfsIndexing = true`
-- `none`: `withCdn = false`, `withIpfsIndexing = false`
-
-Same CID may appear on multiple datasets in one group; dedupe per group when listing or paginating pieces.
-
-Target datasets are looked up with payer + `EARLY_REPAIR_SOURCE` (`utils.ts`, value `early-repair2`).
+All source pieces are deduped globally by CID and repaired into this one target dataset. Target datasets are looked up with target provider + payer + `EARLY_REPAIR_SOURCE` (`utils.ts`) + `withIpfsIndexing = true` + `withCdn = false`.
 
 ## Repair pipeline
 
-### 1. `repair create --provider-id <id>`
+### 1. `repair create --provider-id <id> --target-provider-id <id>`
 
 `createRepair` (`db/create-repair.ts`):
 
-1. **Target provider** — if `--target-provider-id` is set, **`getRepairProvider`** loads that active provider; otherwise **`selectAlternateRepairProvider`** picks one with tier-matched fallback (endorsed → approved → none). Throws if none found.
-2. **`getDataSetsByGroup`** — target provider datasets per group.
-3. Insert **`repairs`** row (`repairProviderId`, `targetProviderId`).
-4. **`forEachPiecesPage`** — paginated `add_piece` operations (`db/get-pieces.ts`, page size 500).
-5. **`getRepairGroups`** — distinct groups from pending `add_piece` ops; saved on the repair row.
-6. For each group with pending pieces but no target dataset → insert **`create_dataset`** operation (`pending`).
+1. **Target provider** — **`getRepairProvider`** loads the required `--target-provider-id`. Throws if none is found.
+2. Insert **`repairs`** row (`repairProviderId`, `targetProviderId`).
+3. **`forEachPiecesPage`** — paginated `add_piece` operations (`db/get-pieces.ts`).
+4. **`getRepairDataset`** — find the target IPFS-enabled dataset for the repair wallet.
+5. If no target dataset exists → insert one **`create_dataset`** operation (`pending`).
 
 ### 2. `repair run <repairId>`
 
-1. Run **`create_dataset`** phase (`pipeline/create-datasets.ts`) via fastq — `SP.createDataSet`, then `updateOperation`; returns created dataset IDs indexed by group.
-2. Run **`add_piece`** pull phase (`pipeline/pull.ts`): pending ops are fetched in same-group pages and fed into fastq as workers free up (`createPullPiecesWorker` — mock logs CIDs per batch).
+1. Run **`create_dataset`** phase (`pipeline/create-datasets.ts`) for the single pending dataset operation — `SP.createDataSet`, then `updateOperation`; stores the single IPFS target dataset ID.
+2. Run **`add_piece`** pull phase (`pipeline/pull.ts`) via `p-queue`: pending ops are fetched in ID order with bounded queue backpressure.
 
-`--reset` retries `pending` and `failed` `create_dataset` ops only; `add_piece` always runs `pending` (failed pieces are skipped). `--batch-size` (default 50) caps pieces per pull job; each batch is one repair group only.
+`--reset` retries `pending` and `failed` `create_dataset` ops and also includes failed `add_piece` ops; otherwise `add_piece` runs only `pending` operations. `--batch-size` caps pieces per pull job.
 
 ### Operation types
 
 - `create_dataset`: `pending` → `committing` → `completed` | `failed`; data has `serviceUrl`, `payee`.
 - `add_piece`: `pending` → `pulling` → `committing` → `completed` | `failed`; data has `cid`, `serviceUrl`, `metadata`, `alternateProviders`.
 
-`add_piece` without alternate providers (other replicas) is created as **`failed`** with error `"No alternate providers found"`. `getProvidersByCid` excludes the source `providerId`.
+`add_piece` without alternate providers (other replicas) is created as **`skipped`** with error `"No alternate providers found"`.
 
 ## Source layout
 
@@ -66,10 +58,8 @@ src/
     wallet.ts
   db/
     create-repair.ts     # createRepair orchestration
-    get-repair-groups.ts   # source groups that need repair
-    get-datasets-by-group.ts # target datasets per group
+    get-repair-dataset.ts # target IPFS-enabled dataset
     get-providers-by-cid.ts  # alternate providers per CID
-    select-alternate-repair-provider.ts # automatic target provider selection
     get-repair-provider.ts             # load explicit target provider by ID
     update-operation.ts      # patch local operation status/result/error
     delete-repair.ts         # delete repair and its operations
@@ -80,7 +70,7 @@ src/
   local-schema.ts        # SQLite repairs/operations
   indexer-schema.ts      # Postgres early-repair schema
   middleware.ts          # DB + wallet context
-  types.ts               # Group, PIECE_GROUPS, DB types
+  types.ts               # Shared DB and context types
   error.ts               # NoAlternateProviderError, RepairCreationError
   utils.ts               # config, client, metadata helpers
 ```
@@ -89,25 +79,21 @@ src/
 
 - Indexer helpers take `IndexerQueryOptions` (`indexerDb`, `indexerSchema`).
 - Extract DB helpers under `src/db/` (indexer queries and local operation updates).
-- Add JSDoc on exported functions/types; inline comments only for non-obvious logic (dedupe, pagination, tier fallback).
-- Use `PIECE_GROUPS` instead of `Object.keys` for group iteration.
+- Add JSDoc on exported functions/types; inline comments only for non-obvious logic (dedupe, pagination).
+- Repairs use one IPFS-enabled target dataset; do not add per-operation dataset grouping.
 
 ## Indexer API (`src/db/`)
 
-| Function                        | Module                              | Purpose                                                 |
-| ------------------------------- | ----------------------------------- | ------------------------------------------------------- |
-| `getRepairGroups`               | `get-repair-groups.ts`              | Repair groups from pending local `add_piece` operations   |
-| `getDataSetsByGroup`            | `get-datasets-by-group.ts`          | One dataset per group for payer + `EARLY_REPAIR_SOURCE` |
-| `getProvidersByCid`             | `get-providers-by-cid.ts`           | Alternate providers per CID; empty array if none        |
-| `selectAlternateRepairProvider` | `select-alternate-repair-provider.ts` | Automatic target provider selection                   |
-| `getRepairProvider`             | `get-repair-provider.ts`              | Load explicit target provider by ID                   |
-| `filterPullPiecesNotInDataset`  | `filter-pull-pieces-not-in-dataset.ts` | Exclude CIDs already indexed in a dataset            |
-| `updateOperation`               | `update-operation.ts`               | Patch local operation status/result/error               |
+- `getRepairDataset` (`get-repair-dataset.ts`): single IPFS-enabled dataset for payer + `EARLY_REPAIR_SOURCE`.
+- `getProvidersByCid` (`get-providers-by-cid.ts`): alternate providers per CID; empty array if none.
+- `getRepairProvider` (`get-repair-provider.ts`): load explicit target provider by ID.
+- `filterPullPiecesNotInDataset` (`filter-pull-pieces-not-in-dataset.ts`): exclude CIDs already indexed in a dataset.
+- `updateOperation` (`update-operation.ts`): patch local operation status/result/error.
 
 ## Local schema
 
-- **`repairs`**: `repairProviderId`, `targetProviderId`, `repairGroups`, `status` (`pending` \| `running` \| `completed` \| `failed`).
-- **`operations`**: `type`, `group`, `status`, `data` (JSON), `result`, `error`.
+- **`repairs`**: `repairProviderId`, `targetProviderId`, `targetDataSetId`, `status` (`pending` \| `completed` \| `failed`).
+- **`operations`**: `type`, `status`, `data` (JSON), `result`, `error`.
 
 ## Commands
 
@@ -118,7 +104,7 @@ src/
 | `repair wallet balance` | Wallet FIL/USDFC balances and pay account summary |
 | `repair wallet deposit <amount>` | Deposit USDFC to pay account |
 | `repair wallet withdraw <amount>` | Withdraw USDFC from pay account |
-| `repair repair create --provider-id <id>` | Plan repair; optional `--target-provider-id`; returns `repairId` |
+| `repair repair create --provider-id <id> --target-provider-id <id>` | Plan repair; returns `repairId` |
 | `repair repair list` | List repairs with operation counts |
 | `repair repair delete <repairId>` | Delete a repair and its operations |
 | `repair repair run <repairId>` | Execute workers; `--concurrency`, `--batch-size`, `--reset` |

packages/repair-cli/package.json

@@ -110,10 +110,13 @@
     "conf": "^15.1.0",
     "drizzle-kit": "^0.31.10",
     "drizzle-orm": "catalog:",
-    "fastq": "^1.20.1",
     "incur": "^0.4.6",
+    "iso-base": "^4.4.0",
     "iso-web": "^2.2.1",
+    "p-all": "^5.0.1",
     "p-locate": "^7.0.0",
+    "p-map": "^7.0.4",
+    "p-queue": "^9.3.0",
     "pg": "^8.21.0",
     "terminal-link": "^5.0.0"
   },

packages/repair-cli/src/commands/repair.ts

@@ -15,7 +15,7 @@ repair.command('create', {
   description: 'Create a new repair',
   options: globalOptions.extend({
     providerId: z.coerce.bigint().describe('Provider ID to repair'),
-    targetProviderId: z.coerce.bigint().optional().describe('Target provider ID for repair'),
+    targetProviderId: z.coerce.bigint().describe('Target provider ID for repair'),
   }),
   middleware: [contextMiddleware],
   run: async (c) => {
@@ -32,6 +32,7 @@ repair.command('create', {
         repairId,
       })
     } catch (error) {
+      console.error(error)
       return c.error({
         code: 'REPAIR_FAILED',
         message: error instanceof Error ? error.message : 'Failed to repair the dataset',
@@ -65,7 +66,8 @@ repair.command('list', {
           repairProviderId: repairWithoutOperations.repairProviderId,
           targetProviderId: repairWithoutOperations.targetProviderId,
           targetProviderUrl: repairWithoutOperations.targetProviderUrl,
-          targetDataSets: repairWithoutOperations.targetDataSets,
+          targetDataSetId: repairWithoutOperations.targetDataSetId,
+          blockNumber: repairWithoutOperations.blockNumber,
           createdAt: new Date(repairWithoutOperations.createdAt).toISOString(),
           updatedAt: new Date(repairWithoutOperations.updatedAt).toISOString(),
           operations: operations.length,
@@ -130,7 +132,7 @@ repair.command('run', {
   }),
   options: globalOptions.extend({
     concurrency: z.coerce.number().default(4).describe('Concurrency level'),
-    batchSize: z.coerce.number().default(10).describe('Max add_piece operations per pull batch (same group)'),
+    batchSize: z.coerce.number().default(10).describe('Max add_piece operations per pull batch'),
     reset: z.boolean().default(false).describe('Reset the repair'),
   }),
   middleware: [contextMiddleware],
@@ -152,7 +154,6 @@ repair.command('run', {
       await runCreateDatasetsPhase({
         ...c.var,
         repair,
-        concurrency,
         reset: c.options.reset,
       })
 

packages/repair-cli/src/db/create-repair.ts

@@ -1,16 +1,14 @@
 import { eq } from 'drizzle-orm'
+import { getBlockNumber } from 'viem/actions'
 import { NoAlternateProviderError, RepairCreationError } from '../error.ts'
-import type { RepairTargetDataSets } from '../local-schema.ts'
 import type { Context } from '../types.ts'
-import { getDataSetsByGroup } from './get-datasets-by-group.ts'
 import { forEachPiecesPage } from './get-pieces.ts'
-import { getRepairGroups } from './get-repair-groups.ts'
+import { getRepairDataset } from './get-repair-dataset.ts'
 import { getRepairProvider } from './get-repair-provider.ts'
-import { selectAlternateRepairProvider } from './select-alternate-repair-provider.ts'
 
 export interface CreateRepairOptions extends Context {
   repairProviderId: bigint
-  targetProviderId?: bigint
+  targetProviderId: bigint
 }
 
 /**
@@ -24,20 +22,16 @@ export async function createRepair(options: CreateRepairOptions): Promise<number
   const { indexerDb, localDb, repairProviderId, targetProviderId, client } = options
   const localSchema = localDb._.fullSchema
   const now = Date.now()
+  const blockNumber = await getBlockNumber(client)
 
-  // Select the target provider
-  if (targetProviderId != null && targetProviderId === repairProviderId) {
+  // Load the explicit target provider.
+  if (targetProviderId === repairProviderId) {
     throw new RepairCreationError('Target provider must differ from the provider being repaired')
   }
-  const targetProvider = targetProviderId
-    ? await getRepairProvider({
-        indexerDb,
-        providerId: targetProviderId,
-      })
-    : await selectAlternateRepairProvider({
-        indexerDb,
-        providerId: repairProviderId,
-      })
+  const targetProvider = await getRepairProvider({
+    indexerDb,
+    providerId: targetProviderId,
+  })
 
   if (!targetProvider) {
     throw new NoAlternateProviderError(targetProviderId)
@@ -50,7 +44,8 @@ export async function createRepair(options: CreateRepairOptions): Promise<number
       repairProviderId,
       targetProviderId: targetProvider.providerId,
       targetProviderUrl: targetProvider.serviceUrl,
-      targetDataSets: {},
+      targetDataSetId: null,
+      blockNumber,
       createdAt: now,
       updatedAt: now,
     })
@@ -59,52 +54,50 @@ export async function createRepair(options: CreateRepairOptions): Promise<number
   if (!repair) throw new RepairCreationError()
 
   // Add the pieces to the repair
+  let hasPendingPieces = false
   await forEachPiecesPage(
     {
       indexerDb,
       providerId: repairProviderId,
       repairId: repair.id,
+      blockNumber,
     },
     async (page) => {
       if (page.operations.length > 0) {
+        hasPendingPieces ||= page.operations.some((operation) => operation.status === 'pending')
         await localDb.insert(localSchema.operations).values(page.operations)
       }
     }
   )
 
-  // Get the repair groups
-  const repairGroups = await getRepairGroups({ localDb, repairId: repair.id })
-  // Get the target datasets for the repair. If no dataset is found, a new one will be created.
-  const dataSetsByGroup = await getDataSetsByGroup({
+  if (!hasPendingPieces) {
+    return repair.id
+  }
+
+  // Get the single IPFS-enabled target dataset for the repair. If none exists, create one before pulling.
+  const targetDataset = await getRepairDataset({
     indexerDb,
     providerId: targetProvider.providerId,
     payer: client.account.address,
+    blockNumber,
   })
 
-  // Each group needs a target dataset before pieces can be added; queue missing datasets first.
-  const targetDataSets: RepairTargetDataSets = {}
-  for (const group of repairGroups) {
-    const dataSet = dataSetsByGroup[group]
-    if (dataSet) {
-      targetDataSets[group] = dataSet.dataSetId
-    } else {
-      targetDataSets[group] = null
-      await localDb.insert(localSchema.operations).values({
-        repairId: repair.id,
-        type: 'create_dataset',
-        group,
-        status: 'pending',
-        data: {
-          payee: targetProvider.providerAddress,
-        },
-        createdAt: now,
-        updatedAt: now,
-      })
-    }
+  if (!targetDataset) {
+    await localDb.insert(localSchema.operations).values({
+      repairId: repair.id,
+      type: 'create_dataset',
+      status: 'pending',
+      data: {
+        payee: targetProvider.providerAddress,
+      },
+      createdAt: now,
+      updatedAt: now,
+    })
   }
+
   await localDb
     .update(localSchema.repairs)
-    .set({ targetDataSets, updatedAt: now })
+    .set({ targetDataSetId: targetDataset?.dataSetId ?? null, updatedAt: now })
     .where(eq(localSchema.repairs.id, repair.id))
 
   return repair.id