feat(09-03): add RouteToUser types and resolver registry

- Add RouteToUser, ResolvedRecipient, NotifyResolver, UserRouteFallback,
  UserRouteFallbackReason types to lib/types/pipeline.ts (ROUTE-01)
- Create lib/services/pipeline-steps/notify-resolvers.ts with three v1
  resolvers: direct_email, pulse_user_id, autotask_resource_email (ROUTE-02)
- Resolver registry as Map<string, NotifyResolver> with registerResolver()
  and resolveRecipient() dispatcher with try/catch error handling

lib/services/pipeline-steps/notify-resolvers.ts

@@ -0,0 +1,103 @@
+/**
+ * Notify Step — user resolver registry.
+ * Resolvers turn a raw PipelineContext field value into a ResolvedRecipient
+ * (a Pulse user_id or an email address). notify.ts calls resolveRecipient()
+ * to delegate to the appropriate resolver.
+ *
+ * Adding a new resolver: call registerResolver(name, fn) — no other files need
+ * to change. All three v1 resolvers are registered at module load below.
+ */
+
+import { postgresClient } from '../postgres-client';
+import { NotifyResolver, ResolvedRecipient, ResolverName } from '../../types/pipeline';
+
+// ============================================================================
+// v1 resolver implementations
+// ============================================================================
+
+/**
+ * direct_email — the field value IS already an email string.
+ * Validates format; returns { email } or null.
+ */
+const directEmail: NotifyResolver = async (fieldValue: unknown): Promise<ResolvedRecipient> => {
+  if (typeof fieldValue !== 'string') return null;
+  if (!/^[^\s@]+@[^\s@]+\.[^\s@]+$/.test(fieldValue)) return null;
+  return { email: fieldValue };
+};
+
+/**
+ * pulse_user_id — the field value IS already a Pulse user.id string.
+ * Checks existence in the "user" table; returns { user_id } or null.
+ */
+const pulseUserId: NotifyResolver = async (fieldValue: unknown): Promise<ResolvedRecipient> => {
+  if (typeof fieldValue !== 'string' || fieldValue === '') return null;
+  const res = await postgresClient.query<{ id: string }>(
+    `SELECT id FROM "user" WHERE id = $1`,
+    [fieldValue],
+  );
+  if (res.rows.length === 0) return null;
+  return { user_id: res.rows[0].id };
+};
+
+/**
+ * autotask_resource_email — the field value is an Autotask resource ID
+ * (integer or numeric string). Looks up the resource's email in the
+ * resources table; returns { email } or null.
+ */
+const autotaskResourceEmail: NotifyResolver = async (fieldValue: unknown): Promise<ResolvedRecipient> => {
+  let resourceId: number;
+  if (typeof fieldValue === 'number') {
+    resourceId = fieldValue;
+  } else if (typeof fieldValue === 'string' && /^\d+$/.test(fieldValue)) {
+    resourceId = parseInt(fieldValue, 10);
+  } else {
+    return null;
+  }
+
+  const res = await postgresClient.query<{ email: string }>(
+    `SELECT email FROM resources WHERE id = $1`,
+    [resourceId],
+  );
+  if (res.rows.length === 0 || !res.rows[0].email) return null;
+  return { email: res.rows[0].email };
+};
+
+// ============================================================================
+// Resolver registry
+// ============================================================================
+
+export const RESOLVERS: Map<string, NotifyResolver> = new Map([
+  ['direct_email', directEmail],
+  ['pulse_user_id', pulseUserId],
+  ['autotask_resource_email', autotaskResourceEmail],
+]);
+
+/**
+ * Register a custom resolver at runtime (e.g., from a plugin or test).
+ * Overwrites any existing resolver with the same name.
+ */
+export function registerResolver(name: ResolverName, fn: NotifyResolver): void {
+  RESOLVERS.set(name, fn);
+}
+
+/**
+ * Resolve a field value to a ResolvedRecipient using the named resolver.
+ * Returns resolverFound=false when the name is not in the registry.
+ * Catches resolver errors and returns recipient=null with resolverFound=true.
+ */
+export async function resolveRecipient(
+  name: ResolverName,
+  fieldValue: unknown,
+): Promise<{ recipient: ResolvedRecipient; resolverFound: boolean }> {
+  const fn = RESOLVERS.get(name);
+  if (!fn) {
+    return { recipient: null, resolverFound: false };
+  }
+  try {
+    const recipient = await fn(fieldValue);
+    return { recipient, resolverFound: true };
+  } catch {
+    // Resolver found but threw — treat as lookup failure, not registry miss
+    return { recipient: null, resolverFound: true };
+  }
+}

lib/types/pipeline.ts

@@ -192,3 +192,55 @@ export interface UserEventSubscription {
   enabled: boolean;
   updated_at: Date;
 }
+
+// ============================================================================
+// Phase 9 — notify step per-user routing (ROUTE-01..06)
+// ============================================================================
+
+/** Resolver name; resolvers live in lib/services/pipeline-steps/notify-resolvers.ts. */
+export type ResolverName = 'autotask_resource_email' | 'direct_email' | 'pulse_user_id' | string;
+
+/** Optional block on a `notify` step's config. When present, the step attempts
+ *  delivery via the resolved Pulse user's personal channel before falling back
+ *  to the step's `channel_id`.
+ *
+ *  v1 limitation: `field` is a single-level key; dotted paths like 'company.id'
+ *  are NOT walked. Admins must surface needed values at top level of the source
+ *  object (e.g., via an upstream transform step). */
+export interface RouteToUser {
+  /** PipelineContext key whose value holds the entity (e.g. 'ticket'). */
+  source: string;
+  /** Single-level field name within context[source] (e.g. 'assignedResourceID').
+   *  v1 only: single-level lookup — dotted paths like 'company.id' are NOT walked. */
+  field: string;
+  /** Resolver name. Looks up the user from the field value. */
+  resolve: ResolverName;
+  /** Event key checked against user_event_subscriptions for muting (D-12). */
+  event_key: string;
+  /** Optional preferred channel type (D-10). When omitted: ntfy then teams then fallback. */
+  channel_type?: 'teams' | 'ntfy';
+}
+
+/** Output of a resolver. null = no recipient (notify.ts falls through to global). */
+export type ResolvedRecipient =
+  | { user_id: string }
+  | { email: string }
+  | null;
+
+/** A resolver fn. Pure async; reads from postgres if it needs to. */
+export type NotifyResolver = (fieldValue: unknown) => Promise<ResolvedRecipient>;
+
+/** Reasons recorded in execution_step.output_data when fallback occurs (D-11). */
+export type UserRouteFallbackReason =
+  | 'no_channel'
+  | 'send_failed'
+  | 'user_not_found'
+  | 'no_field_value'
+  | 'resolver_unknown';
+
+export interface UserRouteFallback {
+  reason: UserRouteFallbackReason;
+  user_id?: string;
+  channel_type: string;
+  error?: string;
+}