fix: mini fixes (#659)

src/bootstrap.ts

@@ -92,7 +92,6 @@ export async function run(
 
     // Push nothing to trigger event
     consumer.push(null as any);
-    processing = false;
   }
 
   consumer.pause();
@@ -159,7 +158,7 @@ function createPkgConsumer(
     try {
       datadog.increment('packages');
 
-      const res = await npm.getDoc(pkg.id);
+      const res = await npm.getDoc(pkg.id, pkg.value.rev);
 
       if (isFailure(res)) {
         log.error('Got an error', res.error);

src/npm/Prefetcher.ts

@@ -1,12 +1,17 @@
-import type { DocumentListParams } from 'nano';
+import type { DocumentListParams, DocumentResponseRow } from 'nano';
 
 import { config } from '../config';
 import { log } from '../utils/log';
 import { wait } from '../utils/wait';
 
+import type { GetPackage } from './types';
+
 import * as npm from './index';
 
-export type PrefetchedPkg = { id: string };
+export type PrefetchedPkg = Pick<
+  DocumentResponseRow<GetPackage>,
+  'id' | 'value'
+>;
 
 export class Prefetcher {
   #limit: number = config.bootstrapConcurrency;

src/npm/index.ts

@@ -84,10 +84,13 @@ async function getChanges(
   return results;
 }
 
-async function getDoc(name: string): Promise<DocumentGetResponse & GetPackage> {
+async function getDoc(
+  name: string,
+  rev: string
+): Promise<DocumentGetResponse & GetPackage> {
   const start = Date.now();
 
-  const doc = await db.get(name);
+  const doc = await db.get(name, { rev });
 
   datadog.timing('npm.getDoc', Date.now() - start);
 

src/watch.ts

@@ -112,19 +112,27 @@ async function loop(
   if (change.deleted) {
     // Delete package directly in index
     // Filter does not support async/await but there is no concurrency issue with this
-    mainIndex.deleteObject(change.id);
-    log.info(`Deleted`, change.id);
-    return;
+    throw new Error('deleted');
   }
 
-  const res = await npm.getDoc(change.id);
+  try {
+    const res = await npm.getDoc(change.id, change.changes[0].rev);
 
-  if (isFailure(res)) {
-    log.error('Got an error', res.error);
-    return;
-  }
+    if (isFailure(res)) {
+      log.error('Got an error', res.error);
+      return;
+    }
 
-  await saveDoc({ row: res, index: mainIndex });
+    await saveDoc({ row: res, index: mainIndex });
+  } catch (e) {
+    // this error can be thrown by us or by nano if:
+    // - we received a change that is not marked as "deleted"
+    // - and the package has since been deleted
+    if (e.message === 'deleted') {
+      mainIndex.deleteObject(change.id);
+      log.info(`deleted`, change.id);
+    }
+  }
 }
 
 /**