fix: progress and new synpase-core

Author: hugomrdias

packages/repair-cli/package.json

@@ -103,16 +103,16 @@
     }
   },
   "dependencies": {
-    "@clack/prompts": "^1.4.0",
+    "@clack/prompts": "^1.5.1",
     "@filoz/repair-db": "workspace:*",
-    "@filoz/synapse-core": "^0.5.2",
+    "@filoz/synapse-core": "^0.6.0",
     "@libsql/client": "^0.17.3",
     "conf": "^15.1.0",
     "drizzle-kit": "^0.31.10",
     "drizzle-orm": "catalog:",
     "incur": "^0.4.6",
     "iso-base": "^4.4.0",
-    "iso-web": "^2.2.1",
+    "iso-web": "^3.1.2",
     "p-all": "^5.0.1",
     "p-locate": "^7.0.0",
     "p-map": "^7.0.4",

packages/repair-cli/src/pipeline/pull.ts

@@ -1,3 +1,4 @@
+import { taskLog } from '@clack/prompts'
 import * as Piece from '@filoz/synapse-core/piece'
 import * as SP from '@filoz/synapse-core/sp'
 import { and, asc, eq, gt, inArray } from 'drizzle-orm'
@@ -6,6 +7,7 @@ import { filterPiecesNotInDataset } from '../db/filter-pull-pieces-not-in-datase
 import { getTargetDataset } from '../db/get-target-dataset.ts'
 import { updateOperation } from '../db/update-operation.ts'
 import { upsertOperations } from '../db/upsert-operations.ts'
+
 import type { AddPieceOperationData, SelectOperation, SelectRepair } from '../local-schema.ts'
 import type { IndexerDatabase, LocalDatabase, WalletClient } from '../types.ts'
 
@@ -30,20 +32,31 @@ export function createPullPiecesWorker({
   indexerDb,
   repair,
   client,
+  state,
+  log,
 }: {
   localDb: LocalDatabase
   indexerDb: IndexerDatabase
   repair: SelectRepair
   client: WalletClient
+  state: {
+    totalBatches: number
+    totalOperations: number
+    completedOperations: number
+    failedOperations: number
+  }
+  log: ReturnType<typeof taskLog>
 }) {
-  return async (batch: PullPiecesBatch) => {
+  return async (batch: PullPiecesBatch, batchNumber: number) => {
+    const spin = log.group(`Batch ${batchNumber}/${state.totalBatches}`)
+    spin.message(`Pull 0 completed, 0 failed`)
     try {
       const dataset = await getTargetDataset({ localDb, repairId: repair.id, client })
       // create pull pieces
       const pullPieces: SP.PullPieceInput[] = []
       for (const operation of batch.operations) {
         const data = operation.data as AddPieceOperationData
-        const pieceCid = Piece.parse(data.cid)
+        const pieceCid = Piece.from(data.cid)
         const sourceUrl = new URL(`/piece/${pieceCid.toString()}`, data.alternateProviders[0]).toString()
         pullPieces.push({ pieceCid, sourceUrl, metadata: data.metadata })
       }
@@ -56,13 +69,14 @@ export function createPullPiecesWorker({
         pieces: pullPieces,
         timeout: 1000 * 60 * 30,
         onStatus: (_status) => {
-          // console.log(`${JSON.stringify(_status)}`)
+          const completed = _status.pieces.filter((piece) => piece.status === 'complete').length
+          const failed = _status.pieces.filter((piece) => piece.status === 'failed').length
+          spin.message(`Pull ${completed} completed, ${failed} failed`)
         },
       })
 
-      // console.log('🚀 ~ createPullPiecesWorker ~ pullResult:', pullResult.pieces)
-
       const completedCids = []
+      const failedCids = []
 
       for (const { pieceCid, status } of pullResult.pieces) {
         const operation = batch.operations.find(
@@ -85,6 +99,7 @@ export function createPullPiecesWorker({
           }
           case 'failed': {
             if (operation) {
+              failedCids.push(pieceCid)
               console.log(`cid ${pieceCid} failed to pull from ${data.alternateProviders[0]}`)
               await updateOperation({
                 localDb,
@@ -101,6 +116,8 @@ export function createPullPiecesWorker({
           }
         }
       }
+      state.completedOperations += completedCids.length
+      state.failedOperations += failedCids.length
 
       // prepare operations for commit
       const notInDatasetCids = await filterPiecesNotInDataset({
@@ -111,20 +128,24 @@ export function createPullPiecesWorker({
       const commitPieces: SP.addPieces.PieceType[] = []
       for (const cid of notInDatasetCids) {
         commitPieces.push({
-          pieceCid: Piece.parse(cid),
+          pieceCid: Piece.from(cid),
           metadata: pullPieces.find(({ pieceCid }) => pieceCid.toString() === cid)?.metadata,
         })
       }
-      console.log(`Pulled ${commitPieces.length} pieces in dataset ${dataset.dataSetId}`)
+      spin.success(
+        `Batch ${batchNumber}/${state.totalBatches} ${completedCids.length} completed, ${failedCids.length} failed`
+      )
       // console.log(commitPieces)
     } catch (error) {
-      console.error(error instanceof Error ? error.message : 'Unknown error')
+      state.failedOperations += batch.operations.length
+      const message = error instanceof Error ? error.message : 'Unknown error'
+      spin.error(`Batch ${batchNumber}/${state.totalBatches} - ${message.replace(/\n/g, ' ')}`)
       await upsertOperations({
         localDb,
         operations: batch.operations.map((operation) => ({
           ...operation,
           status: 'failed',
-          error: error instanceof Error ? error.message : 'Unknown error',
+          error: message,
         })),
       })
     }
@@ -151,6 +172,27 @@ export async function runPullPiecesPhase({
   const pullBatchSize = Math.max(1, batchSize)
   let pullCursor = 0
 
+  const totalOperations = await localDb.$count(
+    localSchema.operations,
+    and(
+      eq(localSchema.operations.repairId, repair.id),
+      eq(localSchema.operations.type, 'add_piece'),
+      inArray(localSchema.operations.status, reset ? ['pending', 'failed'] : ['pending'])
+    )
+  )
+  let batchNumber = 0
+  const state = {
+    totalBatches: Math.ceil(totalOperations / pullBatchSize),
+    totalOperations,
+    completedOperations: 0,
+    failedOperations: 0,
+  }
+
+  const log = taskLog({
+    title: 'Pulling pieces',
+    limit: 1,
+  })
+
   async function getNextPullBatch(): Promise<PullPiecesBatch | null> {
     const operations = await localDb.query.operations.findMany({
       where: and(
@@ -170,15 +212,19 @@ export async function runPullPiecesPhase({
     return { operations }
   }
 
-  const pullPiecesWorker = createPullPiecesWorker({ localDb, indexerDb, repair, client })
+  const pullPiecesWorker = createPullPiecesWorker({ localDb, indexerDb, repair, client, state, log })
   const pullPiecesQueue = new PQueue({ concurrency: pullConcurrency })
 
   while (true) {
     await pullPiecesQueue.onSizeLessThan(pullConcurrency)
     const batch = await getNextPullBatch()
     if (!batch) break
-    pullPiecesQueue.add(() => pullPiecesWorker(batch)).catch(console.error)
+    batchNumber++
+    const currentBatchNumber = batchNumber
+    pullPiecesQueue.add(() => pullPiecesWorker(batch, currentBatchNumber)).catch(console.error)
   }
 
   await pullPiecesQueue.onIdle()
+
+  log.success(`Pulled ${state.completedOperations} pieces, ${state.failedOperations} failed`)
 }

packages/repair-cli/src/utils.ts

@@ -93,6 +93,7 @@ export async function createLocalDatabase(dbPath: string): Promise<LocalDatabase
 }
 
 export async function migrateLocalDatabase(db: LocalDatabase) {
+  // @ts-expect-error - TODO: fix this
   const result = await pushSQLiteSchema(schema, db)
   if (result.hasDataLoss) {
     throw new Error('Data loss detected during migration')