feat(registry): add declarative protocol routing and request-scoped storage hook context across registries

test/test.storage.hooks.ts

@@ -3,7 +3,14 @@ import * as qenv from '@push.rocks/qenv';
 import { RegistryStorage } from '../ts/core/classes.registrystorage.js';
 import type { IStorageConfig } from '../ts/core/interfaces.core.js';
 import type { IStorageHooks, IStorageHookContext } from '../ts/core/interfaces.storage.js';
-import { createTrackingHooks, createQuotaHooks, generateTestRunId } from './helpers/registry.js';
+import {
+  createQuotaHooks,
+  createTestPackument,
+  createTestRegistry,
+  createTestTokens,
+  createTrackingHooks,
+  generateTestRunId,
+} from './helpers/registry.js';
 
 const testQenv = new qenv.Qenv('./', './.nogit');
 
@@ -344,6 +351,140 @@ tap.test('withContext: should clear context even on error', async () => {
   await errorStorage.putObject('test/after-error.txt', Buffer.from('ok'));
 });
 
+tap.test('withContext: should isolate concurrent async operations', async () => {
+  const tracker = createTrackingHooks();
+
+  const concurrentStorage = new RegistryStorage(storageConfig, tracker.hooks);
+  await concurrentStorage.init();
+
+  const bucket = (concurrentStorage as any).bucket;
+  const originalFastPut = bucket.fastPut.bind(bucket);
+  const pendingWrites: Array<() => void> = [];
+  let startedWrites = 0;
+  let waitingWrites = 0;
+  let startedResolve: () => void;
+  let waitingResolve: () => void;
+  const bothWritesStarted = new Promise<void>((resolve) => {
+    startedResolve = resolve;
+  });
+  const bothWritesWaiting = new Promise<void>((resolve) => {
+    waitingResolve = resolve;
+  });
+
+  bucket.fastPut = async (options: any) => {
+    startedWrites += 1;
+    if (startedWrites === 2) {
+      startedResolve();
+    }
+
+    await bothWritesStarted;
+
+    await new Promise<void>((resolve) => {
+      pendingWrites.push(resolve);
+      waitingWrites += 1;
+      if (waitingWrites === 2) {
+        waitingResolve();
+      }
+    });
+
+    return originalFastPut(options);
+  };
+
+  try {
+    const opA = concurrentStorage.withContext(
+      {
+        protocol: 'npm',
+        actor: { userId: 'user-a' },
+        metadata: { packageName: 'package-a' },
+      },
+      async () => {
+        await concurrentStorage.putObject('test/concurrent-a.txt', Buffer.from('a'));
+      }
+    );
+
+    const opB = concurrentStorage.withContext(
+      {
+        protocol: 'npm',
+        actor: { userId: 'user-b' },
+        metadata: { packageName: 'package-b' },
+      },
+      async () => {
+        await concurrentStorage.putObject('test/concurrent-b.txt', Buffer.from('b'));
+      }
+    );
+
+    await bothWritesWaiting;
+
+    pendingWrites[0]!();
+    pendingWrites[1]!();
+
+    await Promise.all([opA, opB]);
+    await new Promise(resolve => setTimeout(resolve, 100));
+  } finally {
+    bucket.fastPut = originalFastPut;
+  }
+
+  const afterPutCalls = tracker.calls.filter(
+    (call) => call.method === 'afterPut' && call.context.key.startsWith('test/concurrent-')
+  );
+  expect(afterPutCalls.length).toEqual(2);
+
+  const callByKey = new Map(afterPutCalls.map((call) => [call.context.key, call]));
+  expect(callByKey.get('test/concurrent-a.txt')?.context.actor?.userId).toEqual('user-a');
+  expect(callByKey.get('test/concurrent-a.txt')?.context.metadata?.packageName).toEqual('package-a');
+  expect(callByKey.get('test/concurrent-b.txt')?.context.actor?.userId).toEqual('user-b');
+  expect(callByKey.get('test/concurrent-b.txt')?.context.metadata?.packageName).toEqual('package-b');
+});
+
+tap.test('request hooks: should receive context during real npm publish requests', async () => {
+  const tracker = createTrackingHooks();
+  const registry = await createTestRegistry({ storageHooks: tracker.hooks });
+
+  try {
+    const tokens = await createTestTokens(registry);
+    const packageName = `hooked-package-${generateTestRunId()}`;
+    const version = '1.0.0';
+    const tarball = Buffer.from('hooked tarball data', 'utf-8');
+    const packument = createTestPackument(packageName, version, tarball);
+
+    const response = await registry.handleRequest({
+      method: 'PUT',
+      path: `/npm/${packageName}`,
+      headers: {
+        Authorization: `Bearer ${tokens.npmToken}`,
+        'Content-Type': 'application/json',
+      },
+      query: {},
+      body: packument,
+    });
+
+    expect(response.status).toEqual(201);
+    await new Promise(resolve => setTimeout(resolve, 100));
+
+    const npmWrites = tracker.calls.filter(
+      (call) => call.method === 'beforePut' && call.context.metadata?.packageName === packageName
+    );
+    expect(npmWrites.length).toBeGreaterThanOrEqual(2);
+
+    const packumentWrite = npmWrites.find(
+      (call) => call.context.key === `npm/packages/${packageName}/index.json`
+    );
+    expect(packumentWrite).toBeTruthy();
+    expect(packumentWrite!.context.protocol).toEqual('npm');
+    expect(packumentWrite!.context.actor?.userId).toEqual(tokens.userId);
+    expect(packumentWrite!.context.metadata?.packageName).toEqual(packageName);
+
+    const tarballWrite = npmWrites.find(
+      (call) => call.context.key.endsWith(`-${version}.tgz`)
+    );
+    expect(tarballWrite).toBeTruthy();
+    expect(tarballWrite!.context.metadata?.packageName).toEqual(packageName);
+    expect(tarballWrite!.context.metadata?.version).toEqual(version);
+  } finally {
+    registry.destroy();
+  }
+});
+
 // ============================================================================
 // Graceful Degradation Tests
 // ============================================================================