Merge pull request #35 from constructive-io/feat/jobs-actor-id-header

feat: thread actor_id through job dispatch pipeline

Author: pyramation

CLAUDE.md

@@ -157,12 +157,11 @@ Useful queries:
 -- Check pending jobs
 SELECT id, task_identifier, attempts, locked_by, last_error FROM app_jobs.jobs ORDER BY id;
 
--- Check database record (needed for job insertion)
-SELECT id, name FROM metaschema_public.database;
+-- Set JWT claims for job context (database_id is read internally by add_job)
+SELECT set_config('jwt.claims.database_id', (SELECT id::text FROM metaschema_public.database LIMIT 1), true);
 
 -- Manually insert a test job
 SELECT * FROM app_jobs.add_job(
-  (SELECT id FROM metaschema_public.database LIMIT 1),
   'simple-email'::text,
   '{"to":"test@example.com","subject":"test","html":"<p>hello</p>"}'::json
 );

job/worker/src/index.ts

@@ -11,6 +11,7 @@ export interface JobRow {
   task_identifier: string;
   payload?: unknown;
   database_id?: string;
+  actor_id?: string;
 }
 
 const log = new Logger('jobs:worker');
@@ -117,6 +118,7 @@ export default class Worker {
     await req(task_identifier, {
       body: payload,
       databaseId: job.database_id,
+      actorId: job.actor_id,
       workerId: this.workerId,
       jobId: job.id
     });

job/worker/src/req.ts

@@ -1,6 +1,7 @@
 import http from 'node:http';
 import https from 'node:https';
 import { URL } from 'node:url';
+
 import {
   getCallbackBaseUrl,
   getJobGatewayConfig,
@@ -30,14 +31,15 @@ const getFunctionUrl = (fn: string): string => {
 
 interface RequestOptions {
   body: unknown;
-  databaseId: string;
+  databaseId?: string;
+  actorId?: string;
   workerId: string;
   jobId: string | number;
 }
 
 const request = (
   fn: string,
-  { body, databaseId, workerId, jobId }: RequestOptions
+  { body, databaseId, actorId, workerId, jobId }: RequestOptions
 ) => {
   const url = getFunctionUrl(fn);
   log.info(`dispatching job`, {
@@ -73,7 +75,8 @@ const request = (
           // these are used by job-worker/job-fn
           'X-Worker-Id': workerId,
           'X-Job-Id': String(jobId),
-          'X-Database-Id': databaseId,
+          ...(databaseId ? { 'X-Database-Id': databaseId } : {}),
+          ...(actorId ? { 'X-Actor-Id': actorId } : {}),
 
           // async HTTP completion callback
           'X-Callback-Url': completeUrl

packages/fn-app/src/index.ts

@@ -13,13 +13,15 @@ type JobContext = {
   workerId: string | undefined;
   jobId: string | undefined;
   databaseId: string | undefined;
+  actorId: string | undefined;
 };
 
 function getHeaders(req: any) {
   return {
     'x-worker-id': req.get('X-Worker-Id'),
     'x-job-id': req.get('X-Job-Id'),
     'x-database-id': req.get('X-Database-Id'),
+    'x-actor-id': req.get('X-Actor-Id'),
     'x-callback-url': req.get('X-Callback-Url')
   };
 }
@@ -99,6 +101,10 @@ const sendJobCallback = async (
     headers['X-Database-Id'] = databaseId;
   }
 
+  if (ctx.actorId) {
+    headers['X-Actor-Id'] = ctx.actorId;
+  }
+
   const body: Record<string, unknown> = {
     status
   };
@@ -168,6 +174,7 @@ const createJobApp = () => {
       'Content-Type': 'application/json',
       'X-Worker-Id': req.get('X-Worker-Id'),
       'X-Database-Id': req.get('X-Database-Id'),
+      'X-Actor-Id': req.get('X-Actor-Id'),
       'X-Job-Id': req.get('X-Job-Id')
     });
     next();
@@ -179,7 +186,8 @@ const createJobApp = () => {
       callbackUrl: req.get('X-Callback-Url'),
       workerId: req.get('X-Worker-Id'),
       jobId: req.get('X-Job-Id'),
-      databaseId: req.get('X-Database-Id')
+      databaseId: req.get('X-Database-Id'),
+      actorId: req.get('X-Actor-Id')
     };
 
     // Store on res.locals so the error middleware can also mark callbacks as sent.

packages/fn-runtime/src/context.ts

@@ -4,6 +4,7 @@ import type { FunctionContext } from './types';
 
 type RequestHeaders = {
   databaseId?: string;
+  actorId?: string;
   workerId?: string;
   jobId?: string;
 };
@@ -15,7 +16,7 @@ export const buildContext = (
   const env = process.env as Record<string, string | undefined>;
   const log = createLogger(options.name || 'fn-runtime');
 
-  const { databaseId, workerId, jobId } = headers;
+  const { databaseId, actorId, workerId, jobId } = headers;
 
   // Create GraphQL clients if databaseId is available and GRAPHQL_URL is set
   let client: FunctionContext['client'];
@@ -45,7 +46,7 @@ export const buildContext = (
   }
 
   return {
-    job: { jobId, workerId, databaseId },
+    job: { jobId, workerId, databaseId, actorId },
     client,
     meta,
     log,

packages/fn-runtime/src/server.ts

@@ -13,6 +13,7 @@ export const createFunctionServer = (
       const context = buildContext(
         {
           databaseId: req.get('X-Database-Id') || req.get('x-database-id') || process.env.DEFAULT_DATABASE_ID,
+          actorId: req.get('X-Actor-Id') || req.get('x-actor-id'),
           workerId: req.get('X-Worker-Id') || req.get('x-worker-id'),
           jobId: req.get('X-Job-Id') || req.get('x-job-id')
         },

packages/fn-runtime/src/types.ts

@@ -10,6 +10,7 @@ export type FunctionContext = {
     jobId?: string;
     workerId?: string;
     databaseId?: string;
+    actorId?: string;
   };
   client: GraphQLClient;
   meta: GraphQLClient;

tests/e2e/utils/jobs.ts

@@ -2,14 +2,16 @@ import { TestClient } from './db';
 
 export interface Job {
   id: string;
-  database_id: string;
+  database_id: string | null;
+  actor_id: string | null;
   queue_name: string | null;
   task_identifier: string;
   payload: Record<string, any>;
   priority: number;
   run_at: Date;
   attempts: number;
   max_attempts: number;
+  is_available: boolean;
   key: string | null;
   last_error: string | null;
   locked_at: Date | null;
@@ -23,7 +25,10 @@ export async function addJob(
   payload: Record<string, any>
 ): Promise<Job> {
   const job = await pg.oneOrNone<Job>(
-    `SELECT * FROM app_jobs.add_job($1::uuid, $2::text, $3::json)`,
+    `WITH _cfg AS (
+      SELECT set_config('jwt.claims.database_id', $1, true)
+    )
+    SELECT j.* FROM _cfg, app_jobs.add_job($2::text, $3::json) AS j`,
     [databaseId, taskIdentifier, JSON.stringify(payload)]
   );
   if (!job) {