refactor(provisioning): TypeScript style improvements from audit

- Import types from @kubernetesjs/ops (ServingKnativeDevV1Service, Secret,
  Namespace, InterwebClient) instead of hand-rolling interfaces
- Define typed DB row interfaces (NamespaceRow, SecretRow,
  FunctionDefinitionRow) — use pool.query<T>() eliminating unsafe casts
- Generic ProvisioningHandler<P, R> type (typed payload/result per handler)
- Pass ComputeModuleLoader through ProvisioningContext from compute-worker
  (shared TTL-cached instance, stop creating per-call)
- Extract mergeAndReplace() into k8s-ops.ts (proper module boundary)
- buildKnativeServiceSpec() accepts KnativeBuilderInput (Pick<FunctionDefinitionRow>)
- DI pattern for K8s client: createK8sClient(url) pure factory + getK8sClient()
  env-reading convenience at the edge
- Decompose seed into provisionNamespaces/syncNamespaceSecrets/provisionFunctions
- K8s error type guards with proper type predicates
- Update tests to match new handler signatures

Author: pyramation

job/compute-worker/src/index.ts

@@ -341,7 +341,7 @@ export default class ComputeWorker {
 
   private async doWorkProvisioning(
     job: ComputeJobRow,
-    handler: (payload: Record<string, unknown>, context: { pool: import('pg').Pool; databaseId: string }) => Promise<Record<string, unknown>>,
+    handler: (payload: Record<string, unknown>, context: import('@constructive-io/provisioning-handlers').ProvisioningContext) => Promise<Record<string, unknown>>,
     payload: Record<string, unknown>
   ): Promise<void> {
     const { task_identifier } = job;
@@ -364,6 +364,7 @@ export default class ComputeWorker {
       const result = await handler(payload, {
         pool: this.pgPool,
         databaseId,
+        loader: this.loader,
       });
 
       const elapsed = process.hrtime(reqStart);

packages/provisioning-handlers/src/handlers/function-sync-resources.ts

@@ -1,93 +1,81 @@
 /**
- * function:sync-resources — updates an existing Knative Service spec
- * when a function definition's resource config changes.
- * Idempotent: replaces the Service spec in-place.
+ * Handler: function:sync-resources
  *
- * Queue handler — triggered by DB trigger on function_definitions UPDATE.
- * Assumes the Knative Service was already created by the seed script.
+ * Queue handler — triggered by DB events on function_definitions UPDATE
+ * (scaling changes, image updates, resource adjustments).
+ *
+ * Reads the updated function definition, rebuilds the Knative Service spec,
+ * and replaces it via the merge-and-replace workflow.
+ *
+ * Assumes the Knative Service already exists (created by the seed).
  */
 
-import { ComputeModuleLoader } from '@constructive-io/module-loader';
 import { Logger } from '@pgpmjs/logger';
 
-import { getK8sClient, isNotFound } from '../k8s-client';
+import { getK8sClient } from '../k8s-client';
+import { mergeAndReplace } from '../k8s-ops';
 import { buildKnativeServiceSpec, resolveNamespaceName } from '../knative';
-import { mergeAndReplace } from '../seed';
-import type { ProvisioningContext, ProvisioningHandler } from '../types';
+import type {
+  FunctionDefinitionRow,
+  ProvisioningHandler,
+  SyncResourcesPayload,
+  SyncResourcesResult,
+} from '../types';
 
-const log = new Logger('provisioning:function-sync');
+const log = new Logger('provisioning:function-sync-resources');
 
-export const handleFunctionSyncResources: ProvisioningHandler = async (
-  payload: Record<string, unknown>,
-  context: ProvisioningContext
-): Promise<Record<string, unknown>> => {
-  const { pool, databaseId } = context;
-  const functionId = payload.id as string;
+export const functionSyncResources: ProvisioningHandler<SyncResourcesPayload, SyncResourcesResult> = async (
+  payload,
+  { pool, databaseId, loader }
+) => {
+  const functionId = payload.id;
 
-  if (!functionId) {
-    throw new Error('function:sync-resources — missing "id" in payload');
+  const client = getK8sClient();
+  if (!client) {
+    log.info('[dev-mode] skipping function:sync-resources (no K8S_API_URL)');
+    return { skipped: true, reason: 'no-k8s' };
   }
 
-  const loader = new ComputeModuleLoader(pool);
+  // Resolve table names via shared module loader (TTL-cached instance)
   const config = await loader.load(databaseId);
   const publicSchema = config.functionModule?.publicSchema ?? 'constructive_compute_public';
   const definitionsTable = config.functionModule?.definitionsTable ?? 'platform_function_definitions';
 
-  const { rows } = await pool.query(
+  const { rows } = await pool.query<FunctionDefinitionRow>(
     `SELECT id, name, task_identifier, service_url, runtime, image,
-            concurrency, scale_min, scale_max, timeout_seconds, resources,
+            concurrency, scale_min, scale_max, scale_target, timeout_seconds, resources,
             namespace_id
      FROM "${publicSchema}"."${definitionsTable}"
      WHERE id = $1`,
     [functionId]
   );
 
   if (rows.length === 0) {
-    throw new Error(`function:sync-resources — function_definition id=${functionId} not found`);
-  }
-
-  const fnRow = rows[0] as Record<string, unknown>;
-
-  if (fnRow.runtime === 'inline' || !fnRow.image) {
-    log.info(
-      `skipping sync for "${fnRow.name}" — ${fnRow.runtime === 'inline' ? 'inline runtime' : 'no image'}`
-    );
-    return { skipped: true, reason: fnRow.runtime === 'inline' ? 'inline-runtime' : 'no-image' };
+    log.warn(`function definition not found: ${functionId}`);
+    return { skipped: true, reason: 'not-found' };
   }
 
-  const client = getK8sClient();
-  if (!client) {
-    log.info(
-      `[dev-mode] would sync Knative Service resources for "${fnRow.name}" — skipping (no K8S_API_URL)`
-    );
-    return { skipped: true, reason: 'no-k8s' };
+  const fnRow = rows[0];
+  if (!fnRow.image || fnRow.runtime === 'inline') {
+    log.info(`skipping inline/image-less function: ${fnRow.name}`);
+    return { skipped: true, reason: 'inline-or-no-image' };
   }
 
-  const namespaceName = await resolveNamespaceName(pool, fnRow.namespace_id as string | null);
-  const fnName = fnRow.name as string;
+  const namespaceName = await resolveNamespaceName(pool, fnRow.namespace_id);
   const serviceSpec = buildKnativeServiceSpec(fnRow, namespaceName);
 
-  try {
-    const svc = await mergeAndReplace(client, serviceSpec, fnName, namespaceName);
-    const serviceUrl = svc?.status?.url ?? svc?.status?.address?.url ?? null;
-    log.info(`updated Knative Service "${fnName}" in namespace "${namespaceName}"`);
-
-    if (serviceUrl && serviceUrl !== fnRow.service_url) {
-      await pool.query(
-        `UPDATE "${publicSchema}"."${definitionsTable}" SET service_url = $1 WHERE id = $2`,
-        [serviceUrl, functionId]
-      );
-      log.info(`updated service_url for "${fnName}" → ${serviceUrl}`);
-    }
+  const svc = await mergeAndReplace(client, serviceSpec, fnRow.name, namespaceName);
+  const serviceUrl = (svc?.status as any)?.url ?? (svc?.status as any)?.address?.url ?? null;
 
-    return { synced: true, name: fnName, serviceUrl };
-  } catch (err: unknown) {
-    if (isNotFound(err)) {
-      log.warn(
-        `Knative Service "${fnName}" not found in "${namespaceName}" — run the provision seed first`
-      );
-      return { skipped: true, reason: 'service-not-found' };
-    }
-    throw err;
+  // Write back updated service_url if available
+  if (serviceUrl && serviceUrl !== fnRow.service_url) {
+    await pool.query(
+      `UPDATE "${publicSchema}"."${definitionsTable}" SET service_url = $1 WHERE id = $2`,
+      [serviceUrl, fnRow.id]
+    );
+    log.info(`updated service_url for "${fnRow.name}" → ${serviceUrl}`);
   }
+
+  log.info(`synced Knative Service "${fnRow.name}" in "${namespaceName}"`);
+  return { synced: true, name: fnRow.name, serviceUrl };
 };

packages/provisioning-handlers/src/handlers/namespace-sync-secrets.ts

@@ -1,55 +1,69 @@
 /**
- * namespace:sync-secrets — reads secrets for a namespace from DB,
- * decrypts them via pgp_sym_decrypt, and syncs to a K8s Secret.
- * Idempotent: creates or replaces the K8s Secret resource.
+ * Handler: namespace:sync-secrets
  *
- * Queue handler — triggered by DB trigger on namespace_secret changes.
+ * Queue handler — triggered by DB events on namespace_secret INSERT/UPDATE/DELETE.
+ * Reads all secrets for a namespace, decrypts them, and creates/replaces
+ * the aggregate K8s Secret in the namespace.
+ *
+ * Assumes the K8s namespace already exists (created by the seed).
  */
 
+import type { Secret } from '@kubernetesjs/ops';
 import { Logger } from '@pgpmjs/logger';
 
 import { getK8sClient, isConflict } from '../k8s-client';
-import type { ProvisioningContext, ProvisioningHandler } from '../types';
+import type {
+  NamespaceRow,
+  ProvisioningHandler,
+  SecretRow,
+  SyncSecretsPayload,
+  SyncSecretsResult,
+} from '../types';
 
-const log = new Logger('provisioning:namespace-secrets');
+const log = new Logger('provisioning:namespace-sync-secrets');
 
-export const handleNamespaceSyncSecrets: ProvisioningHandler = async (
-  payload: Record<string, unknown>,
-  context: ProvisioningContext
-): Promise<Record<string, unknown>> => {
-  const { pool } = context;
-  const namespaceId = payload.id as string | undefined;
-  const namespaceName = payload.namespace_name as string | undefined;
+export const namespaceSyncSecrets: ProvisioningHandler<SyncSecretsPayload, SyncSecretsResult> = async (
+  payload,
+  { pool }
+) => {
+  const namespaceId = payload.id;
+  const namespaceName = payload.namespace_name;
 
-  if (!namespaceId && !namespaceName) {
-    throw new Error('namespace:sync-secrets — missing "id" or "namespace_name" in payload');
+  const client = getK8sClient();
+  if (!client) {
+    log.info('[dev-mode] skipping namespace:sync-secrets (no K8S_API_URL)');
+    return { skipped: true, reason: 'no-k8s' };
   }
 
-  // Resolve namespace name from id if not provided directly
-  let resolvedName = namespaceName;
-  let resolvedId = namespaceId;
-  if (!resolvedName && resolvedId) {
-    const { rows } = await pool.query(
-      `SELECT name FROM metaschema_public.namespace WHERE id = $1`,
-      [resolvedId]
-    );
-    if (rows.length === 0) {
-      throw new Error(`namespace:sync-secrets — namespace id=${resolvedId} not found`);
+  // Resolve namespace name from ID if not provided directly
+  let resolvedName: string;
+  let resolvedId: string | undefined;
+  if (namespaceName) {
+    resolvedName = namespaceName;
+    if (!namespaceId) {
+      const { rows } = await pool.query<NamespaceRow>(
+        `SELECT id, name FROM metaschema_public.namespace WHERE name = $1`,
+        [namespaceName]
+      );
+      if (rows.length === 0) return { skipped: true, reason: 'namespace-not-found' };
+      resolvedId = rows[0].id;
+    } else {
+      resolvedId = namespaceId;
     }
-    resolvedName = rows[0].name as string;
-  }
-  if (!resolvedId && resolvedName) {
-    const { rows } = await pool.query(
-      `SELECT id FROM metaschema_public.namespace WHERE name = $1`,
-      [resolvedName]
+  } else if (namespaceId) {
+    resolvedId = namespaceId;
+    const { rows } = await pool.query<NamespaceRow>(
+      `SELECT id, name FROM metaschema_public.namespace WHERE id = $1`,
+      [namespaceId]
     );
-    if (rows.length > 0) {
-      resolvedId = rows[0].id as string;
-    }
+    if (rows.length === 0) return { skipped: true, reason: 'namespace-not-found' };
+    resolvedName = rows[0].name;
+  } else {
+    return { skipped: true, reason: 'missing-id-and-name' };
   }
 
-  // Read and decrypt secrets for this namespace using pgp_sym_decrypt
-  const { rows: secretRows } = await pool.query(
+  // Fetch and decrypt secrets
+  const { rows: secretRows } = await pool.query<SecretRow>(
     `SELECT key, pgp_sym_decrypt(value, key_id::text) AS decrypted_value
      FROM metaschema_public.namespace_secret
      WHERE namespace_id = $1`,
@@ -58,21 +72,13 @@ export const handleNamespaceSyncSecrets: ProvisioningHandler = async (
 
   const secretData: Record<string, string> = {};
   for (const row of secretRows) {
-    secretData[row.key as string] = Buffer.from(row.decrypted_value as string).toString('base64');
-  }
-
-  const client = getK8sClient();
-  if (!client) {
-    log.info(
-      `[dev-mode] would sync ${secretRows.length} secret(s) for namespace "${resolvedName}" — skipping (no K8S_API_URL)`
-    );
-    return { skipped: true, reason: 'no-k8s' };
+    secretData[row.key] = Buffer.from(row.decrypted_value).toString('base64');
   }
 
   const secretName = `${resolvedName}-secrets`;
-  const secretBody = {
-    apiVersion: 'v1' as const,
-    kind: 'Secret' as const,
+  const secretBody: Secret = {
+    apiVersion: 'v1',
+    kind: 'Secret',
     metadata: { name: secretName },
     data: secretData,
     type: 'Opaque',
@@ -81,19 +87,18 @@ export const handleNamespaceSyncSecrets: ProvisioningHandler = async (
   try {
     await client.createCoreV1NamespacedSecret({
       query: {},
-      path: { namespace: resolvedName! },
+      path: { namespace: resolvedName },
       body: secretBody,
     });
-    log.info(`created K8s secret "${secretName}" in namespace "${resolvedName}" with ${secretRows.length} key(s)`);
+    log.info(`created K8s secret "${secretName}" with ${secretRows.length} key(s)`);
   } catch (err: unknown) {
     if (isConflict(err)) {
-      log.info(`K8s secret "${secretName}" already exists — replacing`);
       await client.replaceCoreV1NamespacedSecret({
         query: {},
-        path: { name: secretName, namespace: resolvedName! },
+        path: { name: secretName, namespace: resolvedName },
         body: secretBody,
       });
-      log.info(`replaced K8s secret "${secretName}" in namespace "${resolvedName}"`);
+      log.info(`replaced K8s secret "${secretName}" with ${secretRows.length} key(s)`);
     } else {
       throw err;
     }

packages/provisioning-handlers/src/index.ts

@@ -1,10 +1,30 @@
-export type { ProvisioningContext, ProvisioningHandler } from './types';
+// Types
+export type {
+  FunctionDefinitionRow,
+  NamespaceRow,
+  ProvisioningContext,
+  ProvisioningHandler,
+  SecretRow,
+  SyncResourcesPayload,
+  SyncResourcesResult,
+  SyncSecretsPayload,
+  SyncSecretsResult,
+} from './types';
+
+// Registry
 export {
   getProvisioningHandler,
   registerProvisioningHandler,
 } from './registry';
-export { getK8sClient, isConflict, isNotFound } from './k8s-client';
-export type { KnativeServiceSpec } from './knative';
+
+// K8s client + utilities
+export { createK8sClient, getK8sClient, isConflict, isNotFound } from './k8s-client';
+export { mergeAndReplace } from './k8s-ops';
+
+// Knative builder
+export type { KnativeBuilderInput, KnativeServiceSpec } from './knative';
 export { buildKnativeServiceSpec, resolveNamespaceName } from './knative';
+
+// Seed
 export type { ProvisionSeedOptions, ProvisionSeedResult } from './seed';
-export { mergeAndReplace, provision } from './seed';
+export { provision } from './seed';