Merge remote-tracking branch 'origin/main' into feat/portable-functions

Author: Anmol1696

.github/workflows/test-k8s-deployment.yaml

@@ -139,6 +139,11 @@ jobs:
           kubectl wait --for=condition=complete job/constructive-db \
             -n constructive-functions --timeout=180s
 
+      - name: Restart job service after DB ready
+        run: |
+          kubectl rollout restart deploy/knative-job-service -n constructive-functions
+          kubectl rollout status deploy/knative-job-service -n constructive-functions --timeout=120s
+
       - name: Port-forward postgres and run e2e tests
         env:
           PGHOST: localhost

AGENTS.md

@@ -68,7 +68,7 @@ const handler: FunctionHandler = async (params, context) => {
   // client/meta: GraphQL clients (tenant-scoped, created per-request)
   // log: structured logger
   // env: process.env
-  // job: { jobId, workerId, databaseId }
+  // job: { jobId, workerId, databaseId, actorId }
   return { complete: true };
 };
 

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
 );

functions/send-email-link/handler.ts

@@ -6,37 +6,50 @@ import { send as sendPostmaster } from '@constructive-io/postmaster';
 import { send as sendSmtp } from 'simple-smtp-server';
 import { parseEnvBoolean } from '@pgpmjs/env';
 
+// Use plural connections with the `where` filter (graphile-connection-filter)
+// rather than `condition:` or singular-by-PK. The constructive server's
+// ConstructivePreset:
+//   1. Extends NoUniqueLookupPreset — disables singular root-field lookups
+//      by unique constraint, including primary key (`database(id:)`,
+//      `user(id:)`).
+//   2. Disables `PgConditionArgumentPlugin` — removes the built-in
+//      `condition:` argument on connections.
+// The supported pattern is `where: { field: { equalTo: var } }`.
 const GetUser = gql`
   query GetUser($userId: UUID!) {
-    user(id: $userId) {
-      username
-      displayName
-      profilePicture
+    users(where: { id: { equalTo: $userId } }, first: 1) {
+      nodes {
+        username
+        displayName
+        profilePicture
+      }
     }
   }
 `;
 
 const GetDatabaseInfo = gql`
   query GetDatabaseInfo($databaseId: UUID!) {
-    database(id: $databaseId) {
-      sites {
-        nodes {
-          domains {
-            nodes {
-              subdomain
-              domain
+    databases(where: { id: { equalTo: $databaseId } }, first: 1) {
+      nodes {
+        sites {
+          nodes {
+            domains {
+              nodes {
+                subdomain
+                domain
+              }
             }
-          }
-          logo
-          title
-          siteThemes {
-            nodes {
-              theme
+            logo
+            title
+            siteThemes {
+              nodes {
+                theme
+              }
             }
-          }
-          siteModules(condition: { name: "legal_terms_module" }) {
-            nodes {
-              data
+            siteModules(where: { name: { equalTo: "legal_terms_module" } }) {
+              nodes {
+                data
+              }
             }
           }
         }
@@ -111,7 +124,7 @@ const sendEmailLink = async (
     databaseId
   });
 
-  const site = databaseInfo?.database?.sites?.nodes?.[0];
+  const site = databaseInfo?.databases?.nodes?.[0]?.sites?.nodes?.[0];
   if (!site) {
     throw new Error('Site not found for database');
   }
@@ -177,7 +190,7 @@ const sendEmailLink = async (
       const inviter = await client.request<any>(GetUser, {
         userId: params.sender_id
       });
-      inviterName = inviter?.user?.displayName;
+      inviterName = inviter?.users?.nodes?.[0]?.displayName;
 
       if (inviterName) {
         subject = `${inviterName} invited you to ${nick}!`;

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 { createClients } from './graphql';
 
 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/graphql.ts

@@ -61,11 +61,14 @@ export const createClients = (
     ...(schemata && { schemata })
   });
 
+  // Meta client targets platform-level data (databases, sites, domains, themes
+  // in metaschema_public). It must NOT use tenant API routing — sending
+  // X-Api-Name causes the server to load every schema registered for that API
+  // (both *_public and *_private variants), which collides on duplicate codec
+  // names like identityProviders. Use X-Meta-Schema instead.
   const meta = createGraphQLClient(metaGraphqlUrl, env, {
     hostHeaderEnvVar: 'META_GRAPHQL_HOST_HEADER',
-    databaseId,
-    ...(apiName && { apiName }),
-    ...(schemata && { schemata })
+    useMetaSchema: true,
   });
 
   return { client, meta };

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')
         },