Request context propagation across HTTP, jobs, and DB layer (#213)

packages/backend/src/context/bullmq-context.test.ts

@@ -0,0 +1,82 @@
+import { describe, it, expect, vi } from 'vitest';
+import { context, withContext } from '@logtide/shared/context';
+import { attachContextToPayload, wrapProcessorWithContext, CTX_KEY } from './bullmq-context.js';
+
+describe('attachContextToPayload', () => {
+  it('returns input unchanged when no context', () => {
+    const data = { foo: 1 };
+    expect(attachContextToPayload(data)).toBe(data);
+  });
+
+  it('adds _ctx with v=1 when a context is active', async () => {
+    await withContext({ requestId: 'req-1', organizationId: 'org-1' }, async () => {
+      const out = attachContextToPayload({ foo: 1 }) as any;
+      expect(out._ctx.v).toBe(1);
+      expect(out._ctx.requestId).toBe('req-1');
+      expect(out._ctx.organizationId).toBe('org-1');
+      expect(out.foo).toBe(1);
+    });
+  });
+
+  it('handles non-object payloads gracefully', () => {
+    expect(attachContextToPayload('hello' as unknown as object)).toBe('hello');
+  });
+});
+
+describe('wrapProcessorWithContext', () => {
+  it('strips _ctx and runs processor inside context.run', async () => {
+    const processor = vi.fn(async (job) => {
+      // expect data does not contain _ctx
+      expect((job.data as any)._ctx).toBeUndefined();
+      // expect context is established
+      const ctx = context.current();
+      expect(ctx.requestId).toBe('req-7');
+      expect(ctx.origin).toBe('job');
+    });
+
+    const wrapped = wrapProcessorWithContext('test-job', processor);
+    await wrapped({
+      id: 'j1',
+      name: 'test-job',
+      data: {
+        payload: 'x',
+        [CTX_KEY]: {
+          v: 1,
+          requestId: 'req-7',
+          origin: 'http',
+          actor: { type: 'system', id: null },
+          organizationId: null,
+          projectId: null,
+        },
+      } as any,
+    });
+    expect(processor).toHaveBeenCalledOnce();
+  });
+
+  it('falls back to runAsSystem when _ctx is missing', async () => {
+    const processor = vi.fn(async () => {
+      const ctx = context.current();
+      expect(ctx.origin).toBe('system');
+      expect(ctx.systemReason).toBe('bullmq-job:no-ctx-job');
+    });
+
+    const wrapped = wrapProcessorWithContext('no-ctx-job', processor);
+    await wrapped({ id: 'j2', name: 'no-ctx-job', data: { payload: 'y' } as any });
+    expect(processor).toHaveBeenCalledOnce();
+  });
+
+  it('falls back to runAsSystem when _ctx has unknown version', async () => {
+    const processor = vi.fn(async () => {
+      const ctx = context.current();
+      expect(ctx.origin).toBe('system');
+    });
+
+    const wrapped = wrapProcessorWithContext('bad-ctx-job', processor);
+    await wrapped({
+      id: 'j3',
+      name: 'bad-ctx-job',
+      data: { [CTX_KEY]: { v: 99 } } as any,
+    });
+    expect(processor).toHaveBeenCalledOnce();
+  });
+});

packages/backend/src/context/bullmq-context.ts

@@ -0,0 +1,45 @@
+import { currentOrNull, deserializeContext, run, runAsSystem, serializeContext } from '@logtide/shared/context';
+import type { JobProcessor } from '../queue/abstractions/types.js';
+
+export const CTX_KEY = '_ctx';
+
+/**
+ * Adds `_ctx` to a job payload if a context is currently active.
+ * Returns the (possibly modified) payload as `unknown` to keep adapter generics sane.
+ */
+export function attachContextToPayload<T>(data: T): T {
+  const ctx = currentOrNull();
+  if (!ctx) return data;
+  if (typeof data !== 'object' || data === null) return data;
+  return { ...(data as object), [CTX_KEY]: serializeContext(ctx) } as T;
+}
+
+/**
+ * Wraps a JobProcessor: detaches `_ctx` from the payload (so the user processor sees clean data)
+ * and runs the processor inside the appropriate context.run / runAsSystem.
+ */
+export function wrapProcessorWithContext<T>(
+  jobName: string,
+  processor: JobProcessor<T>
+): JobProcessor<T> {
+  return async (job) => {
+    const raw = job.data as T & { [CTX_KEY]?: unknown };
+    const serialized = (raw as Record<string, unknown> | null)?.[CTX_KEY];
+    const cleanData =
+      raw && typeof raw === 'object'
+        ? (() => {
+            const { [CTX_KEY]: _omit, ...rest } = raw as Record<string, unknown>;
+            return rest as T;
+          })()
+        : raw;
+
+    const cleanJob = { ...job, data: cleanData };
+
+    const ctx = serialized != null ? deserializeContext(serialized) : undefined;
+    if (ctx) {
+      await run(ctx, () => processor(cleanJob));
+    } else {
+      await runAsSystem(`bullmq-job:${jobName}`, () => processor(cleanJob));
+    }
+  };
+}

packages/backend/src/context/database-comment.test.ts

@@ -0,0 +1,36 @@
+import { describe, it, expect } from 'vitest';
+import { withContext, currentOrNull } from '@logtide/shared/context';
+import { formatContextComment } from './kysely-plugin.js';
+
+describe('SQL comment injection helpers', () => {
+  it('formatContextComment can be invoked deterministically', () => {
+    const out = formatContextComment({
+      requestId: 'r1',
+      origin: 'http',
+      organizationId: 'o1',
+      actor: { type: 'user', id: 'u1' },
+    });
+    expect(out).toMatch(/^\/\* req=r1 origin=http org=o1 actor=user:u1 \*\/ $/);
+  });
+
+  it('manually-built patched query function prepends comment', async () => {
+    const captured: string[] = [];
+    const fakeOriginal = (sql: string) => {
+      captured.push(sql);
+      return Promise.resolve({ rows: [], rowCount: 0 });
+    };
+    const patched = (...args: any[]) => {
+      const ctx = currentOrNull();
+      if (!ctx) return fakeOriginal(...(args as [string]));
+      const comment = formatContextComment(ctx);
+      args[0] = comment + args[0];
+      return fakeOriginal(...(args as [string]));
+    };
+
+    await withContext({ requestId: 'rZ', organizationId: 'oZ' }, async () => {
+      await patched('SELECT 1');
+    });
+
+    expect(captured[0]).toMatch(/^\/\* req=rZ /);
+  });
+});

packages/backend/src/context/fastify-plugin.test.ts

@@ -0,0 +1,67 @@
+import { describe, it, expect } from 'vitest';
+import Fastify from 'fastify';
+import { context } from '@logtide/shared/context';
+import { contextPlugin } from './fastify-plugin.js';
+
+describe('contextPlugin', () => {
+  async function buildApp(authStub?: (req: any) => void) {
+    const app = Fastify({ logger: false });
+    if (authStub) {
+      // Simulate the auth plugin running BEFORE contextPlugin (registration order)
+      app.addHook('onRequest', async (req) => {
+        authStub(req);
+      });
+    }
+    await app.register(contextPlugin);
+    app.get('/echo', async () => {
+      const ctx = context.current();
+      return {
+        requestId: ctx.requestId,
+        origin: ctx.origin,
+        actorType: ctx.actor.type,
+        actorId: ctx.actor.id,
+        organizationId: ctx.organizationId,
+      };
+    });
+    return app;
+  }
+
+  it('establishes a context for every request (anonymous)', async () => {
+    const app = await buildApp();
+    const res = await app.inject({ method: 'GET', url: '/echo' });
+    expect(res.statusCode).toBe(200);
+    const body = res.json();
+    expect(body.origin).toBe('http');
+    expect(body.actorType).toBe('system');
+    expect(body.requestId).toBeTruthy();
+    await app.close();
+  });
+
+  it('captures user actor when auth populated request.user', async () => {
+    const app = await buildApp((req) => {
+      req.user = { id: 'u-abc', email: 'alice@test' };
+      req.organizationId = 'org-1';
+    });
+    const res = await app.inject({ method: 'GET', url: '/echo' });
+    const body = res.json();
+    expect(body.actorType).toBe('user');
+    expect(body.actorId).toBe('u-abc');
+    expect(body.organizationId).toBe('org-1');
+    await app.close();
+  });
+
+  it('captures apiKey actor when auth populated request.apiKeyId', async () => {
+    const app = await buildApp((req) => {
+      req.apiKeyId = 'key-xyz';
+      req.apiKeyType = 'write';
+      req.organizationId = 'org-2';
+      req.projectId = 'proj-2';
+    });
+    const res = await app.inject({ method: 'GET', url: '/echo' });
+    const body = res.json();
+    expect(body.actorType).toBe('apiKey');
+    expect(body.actorId).toBe('key-xyz');
+    expect(body.organizationId).toBe('org-2');
+    await app.close();
+  });
+});

packages/backend/src/context/fastify-plugin.ts

@@ -0,0 +1,44 @@
+import type { FastifyPluginAsync, FastifyRequest } from 'fastify';
+import fp from 'fastify-plugin';
+import type { ApiKeyType } from '@logtide/shared';
+import { context, type Actor, type RequestContext } from '@logtide/shared/context';
+
+type AuthDecoratedRequest = FastifyRequest & {
+  organizationId?: string;
+  projectId?: string;
+  apiKeyId?: string;
+  apiKeyType?: ApiKeyType;
+  user?: { id: string; email: string };
+};
+
+const plugin: FastifyPluginAsync = async (fastify) => {
+  fastify.addHook('onRequest', async (request) => {
+    const r = request as AuthDecoratedRequest;
+
+    const actor: Actor = r.user
+      ? { type: 'user', id: r.user.id, email: r.user.email }
+      : r.apiKeyId
+      ? { type: 'apiKey', id: r.apiKeyId, apiKeyType: r.apiKeyType }
+      : { type: 'system', id: null };
+
+    const userAgentHeader = r.headers['user-agent'];
+    const userAgent =
+      typeof userAgentHeader === 'string' ? userAgentHeader : undefined;
+
+    const ctx: RequestContext = {
+      requestId: r.id,
+      origin: 'http',
+      actor,
+      organizationId: r.organizationId ?? null,
+      projectId: r.projectId ?? null,
+      ip: r.ip,
+      userAgent,
+    };
+    context.enterWith(ctx);
+  });
+};
+
+export const contextPlugin = fp(plugin, {
+  name: 'context',
+  fastify: '5.x',
+});

packages/backend/src/context/index.ts

@@ -0,0 +1,2 @@
+export { contextPlugin } from './fastify-plugin.js';
+export { context } from '@logtide/shared/context';

packages/backend/src/context/kysely-plugin.test.ts

@@ -0,0 +1,34 @@
+import { describe, it, expect } from 'vitest';
+import { formatContextComment, safeForComment } from './kysely-plugin.js';
+
+describe('formatContextComment', () => {
+  it('formats basic context', () => {
+    const out = formatContextComment({
+      requestId: 'req-1',
+      origin: 'http',
+      organizationId: 'org-1',
+      actor: { type: 'user', id: 'u1' },
+    });
+    expect(out).toBe('/* req=req-1 origin=http org=org-1 actor=user:u1 */ ');
+  });
+
+  it('uses dash placeholder for null fields', () => {
+    const out = formatContextComment({
+      requestId: 'req-1',
+      origin: 'system',
+      organizationId: null,
+      actor: { type: 'system', id: null },
+    });
+    expect(out).toBe('/* req=req-1 origin=system org=- actor=system:- */ ');
+  });
+});
+
+describe('safeForComment', () => {
+  it('strips dangerous characters', () => {
+    expect(safeForComment('a*/b\nc')).toBe('abc');
+  });
+  it('returns dash for empty result', () => {
+    expect(safeForComment('***')).toBe('-');
+    expect(safeForComment(null)).toBe('-');
+  });
+});

packages/backend/src/context/kysely-plugin.ts

@@ -0,0 +1,45 @@
+import type {
+  KyselyPlugin,
+  PluginTransformQueryArgs,
+  PluginTransformResultArgs,
+  QueryResult,
+  RootOperationNode,
+  UnknownRow,
+} from 'kysely';
+
+/**
+ * v1 placeholder. The actual SQL comment injection happens at the pg.Pool level
+ * (see database/connection.ts) so the comment lands in front of the wire query
+ * without depending on Kysely AST internals (which differ per query kind).
+ *
+ * Kept as a stub so future work can move comment injection back into the plugin
+ * if Kysely exposes a stable RawNode-prefix API.
+ */
+export class ContextSqlCommentPlugin implements KyselyPlugin {
+  transformQuery(args: PluginTransformQueryArgs): RootOperationNode {
+    return args.node;
+  }
+  transformResult(args: PluginTransformResultArgs): Promise<QueryResult<UnknownRow>> {
+    return Promise.resolve(args.result);
+  }
+}
+
+const SAFE_RE = /[^a-zA-Z0-9_:-]/g;
+
+export function safeForComment(value: string | null | undefined): string {
+  if (!value) return '-';
+  const cleaned = value.replace(SAFE_RE, '');
+  return cleaned.length > 0 ? cleaned : '-';
+}
+
+export function formatContextComment(ctx: {
+  requestId: string;
+  origin: string;
+  organizationId: string | null;
+  actor: { type: string; id: string | null };
+}): string {
+  const actor = `${safeForComment(ctx.actor.type)}:${safeForComment(ctx.actor.id)}`;
+  return `/* req=${safeForComment(ctx.requestId)} origin=${safeForComment(
+    ctx.origin
+  )} org=${safeForComment(ctx.organizationId)} actor=${actor} */ `;
+}