Establish complete PostgreSQL request context across Graphile execution

graphile/graphile-bulk-mutations/src/__tests__/pg-client.test.ts

@@ -0,0 +1,35 @@
+import { queryPgClient } from '../utils/pg-client';
+
+describe('queryPgClient', () => {
+  it('uses the native @dataplan/pg query-config contract', async () => {
+    const query = jest.fn(async () => ({ rows: [{ id: 1 }], rowCount: 1 }));
+    const client = { query };
+
+    await expect(
+      queryPgClient<{ id: number }>(
+        client as never,
+        'UPDATE app.items SET name = $1 RETURNING id',
+        ['updated']
+      )
+    ).resolves.toEqual({ rows: [{ id: 1 }], rowCount: 1 });
+
+    expect(query).toHaveBeenCalledTimes(1);
+    expect(query).toHaveBeenCalledWith({
+      text: 'UPDATE app.items SET name = $1 RETURNING id',
+      values: ['updated'],
+    });
+  });
+
+  it('preserves query failures', async () => {
+    const original = new Error('database rejected mutation');
+    const client = {
+      query: jest.fn(async () => {
+        throw original;
+      }),
+    };
+
+    await expect(
+      queryPgClient(client as never, 'DELETE FROM app.items', [])
+    ).rejects.toBe(original);
+  });
+});

graphile/graphile-bulk-mutations/src/plugins/BulkDeletePlugin.ts

@@ -1,9 +1,11 @@
 import '../augmentations';
 
-import { sideEffectWithPgClient } from '@dataplan/pg';
+import { type PgClient, sideEffectWithPgClient } from '@dataplan/pg';
 import type { GraphileConfig } from 'graphile-config';
 import type { GraphQLInputType,GraphQLOutputType } from 'graphql';
 
+import { queryPgClient } from '../utils/pg-client';
+
 const version = '0.1.0';
 
 /**
@@ -105,7 +107,7 @@ export const BulkDeletePlugin: GraphileConfig.Plugin = {
                     const $result = sideEffectWithPgClient(
                       executor,
                       $input,
-                      async (pgClient: any, input: any) => {
+                      async (pgClient: PgClient, input: any) => {
                         if (requireWhere && (!input.where || Object.keys(input.where).length === 0)) {
                           throw new Error(
                             'Bulk delete requires a non-empty where condition. Set bulkRequireWhere: false to allow unrestricted deletes.'
@@ -194,12 +196,16 @@ export const BulkDeletePlugin: GraphileConfig.Plugin = {
                         // Use RETURNING <pk_columns> instead of RETURNING *
                         // For delete, we capture PKs before rows are gone
                         const text = `DELETE FROM ${compiledFrom}\nWHERE ${whereStr}\nRETURNING ${pkReturning}`;
-                        const mutationResult = await pgClient.query(text, values);
+                        const mutationResult = await queryPgClient<Record<string, unknown>>(
+                          pgClient,
+                          text,
+                          values
+                        );
                         const affectedCount = mutationResult.rowCount ?? 0;
 
                         // For delete, rows no longer exist so we can't do a
                         // follow-up SELECT. Return the PK values directly.
-                        const returning = mutationResult.rows || [];
+                        const returning = [...mutationResult.rows];
 
                         return {
                           affectedCount,

graphile/graphile-bulk-mutations/src/plugins/BulkInsertPlugin.ts

@@ -1,9 +1,10 @@
 import '../augmentations';
 
-import { sideEffectWithPgClient } from '@dataplan/pg';
+import { type PgClient, sideEffectWithPgClient } from '@dataplan/pg';
 import type { GraphileConfig } from 'graphile-config';
 import type { GraphQLInputType, GraphQLOutputType } from 'graphql';
 
+import { queryPgClient } from '../utils/pg-client';
 import type { NestedRelationInfo } from '../utils/relations';
 import { discoverNestedRelations } from '../utils/relations';
 import type { ColumnSpec } from '../utils/sql-builder';
@@ -131,7 +132,7 @@ export const BulkInsertPlugin: GraphileConfig.Plugin = {
                     const $result = sideEffectWithPgClient(
                       executor,
                       $input,
-                      async (pgClient: any, input: any) => {
+                      async (pgClient: PgClient, input: any) => {
                         const values = input.values;
                         if (!values || !Array.isArray(values) || values.length === 0) {
                           return { affectedCount: 0, returning: [] };
@@ -200,10 +201,9 @@ export const BulkInsertPlugin: GraphileConfig.Plugin = {
                         const allPkRows: Record<string, unknown>[] = [];
 
                         for (const batch of batches) {
-                          const result = await pgClient.query(
-                            batch.text,
-                            batch.values
-                          );
+                          const result = await queryPgClient<
+                            Record<string, unknown>
+                          >(pgClient, batch.text, batch.values);
                           totalAffected += result.rowCount ?? 0;
                           if (result.rows) {
                             allPkRows.push(...result.rows);
@@ -259,7 +259,8 @@ export const BulkInsertPlugin: GraphileConfig.Plugin = {
                               );
 
                               for (const batch of childBatches) {
-                                const result = await pgClient.query(
+                                const result = await queryPgClient<unknown>(
+                                  pgClient,
                                   batch.text,
                                   batch.values
                                 );
@@ -282,11 +283,12 @@ export const BulkInsertPlugin: GraphileConfig.Plugin = {
                           const selectParams = allPkRows.flatMap((pkRow) =>
                             pkColumns.map((col) => pkRow[col])
                           );
-                          const selectResult = await pgClient.query(
+                          const selectResult = await queryPgClient<unknown>(
+                            pgClient,
                             `SELECT * FROM ${compiledFrom} WHERE ${whereClause}`,
                             selectParams
                           );
-                          returning = selectResult.rows || [];
+                          returning = [...selectResult.rows];
                         }
 
                         return {

graphile/graphile-bulk-mutations/src/plugins/BulkUpdatePlugin.ts

@@ -1,9 +1,11 @@
 import '../augmentations';
 
-import { sideEffectWithPgClient } from '@dataplan/pg';
+import { type PgClient, sideEffectWithPgClient } from '@dataplan/pg';
 import type { GraphileConfig } from 'graphile-config';
 import type { GraphQLInputType, GraphQLOutputType } from 'graphql';
 
+import { queryPgClient } from '../utils/pg-client';
+
 const version = '0.1.0';
 
 /**
@@ -106,7 +108,7 @@ export const BulkUpdatePlugin: GraphileConfig.Plugin = {
                     const $result = sideEffectWithPgClient(
                       executor,
                       $input,
-                      async (pgClient: any, input: any) => {
+                      async (pgClient: PgClient, input: any) => {
                         if (requireWhere && (!input.where || Object.keys(input.where).length === 0)) {
                           throw new Error(
                             'Bulk update requires a non-empty where condition. Set bulkRequireWhere: false to allow unrestricted updates.'
@@ -212,13 +214,15 @@ export const BulkUpdatePlugin: GraphileConfig.Plugin = {
 
                         // Use RETURNING <pk_columns> instead of RETURNING *
                         const text = `UPDATE ${compiledFrom}\nSET ${setClauses.join(', ')}\nWHERE ${whereStr}\nRETURNING ${pkReturning}`;
-                        const mutationResult = await pgClient.query(text, values);
+                        const mutationResult = await queryPgClient<
+                          Record<string, unknown>
+                        >(pgClient, text, values);
                         const affectedCount = mutationResult.rowCount ?? 0;
 
                         // Follow-up SELECT using PKs to respect column-level grants
                         let returning: unknown[] = [];
                         if (mutationResult.rows && mutationResult.rows.length > 0) {
-                          const pkRows: Record<string, unknown>[] = mutationResult.rows;
+                          const pkRows = mutationResult.rows;
                           const pkConditions = pkRows.map((pkRow, rowIdx) => {
                             return pkColumns.map((col, colIdx) => {
                               const paramIdx = rowIdx * pkColumns.length + colIdx + 1;
@@ -229,11 +233,12 @@ export const BulkUpdatePlugin: GraphileConfig.Plugin = {
                           const selectParams = pkRows.flatMap((pkRow) =>
                             pkColumns.map((col) => pkRow[col])
                           );
-                          const selectResult = await pgClient.query(
+                          const selectResult = await queryPgClient<unknown>(
+                            pgClient,
                             `SELECT * FROM ${compiledFrom} WHERE ${selectWhere}`,
                             selectParams
                           );
-                          returning = selectResult.rows || [];
+                          returning = [...selectResult.rows];
                         }
 
                         return {

graphile/graphile-bulk-mutations/src/plugins/BulkUpsertPlugin.ts

@@ -1,9 +1,10 @@
 import '../augmentations';
 
-import { sideEffectWithPgClient } from '@dataplan/pg';
+import { type PgClient, sideEffectWithPgClient } from '@dataplan/pg';
 import type { GraphileConfig } from 'graphile-config';
 import type { GraphQLInputType, GraphQLOutputType } from 'graphql';
 
+import { queryPgClient } from '../utils/pg-client';
 import type { ColumnSpec } from '../utils/sql-builder';
 import { buildBulkInsertSQL } from '../utils/sql-builder';
 
@@ -118,7 +119,7 @@ export const BulkUpsertPlugin: GraphileConfig.Plugin = {
                     const $result = sideEffectWithPgClient(
                       executor,
                       $input,
-                      async (pgClient: any, input: any) => {
+                      async (pgClient: PgClient, input: any) => {
                         const values = input.values;
                         if (!values || !Array.isArray(values) || values.length === 0) {
                           return { affectedCount: 0, returning: [] };
@@ -177,10 +178,9 @@ export const BulkUpsertPlugin: GraphileConfig.Plugin = {
                         const allPkRows: Record<string, unknown>[] = [];
 
                         for (const batch of batches) {
-                          const result = await pgClient.query(
-                            batch.text,
-                            batch.values
-                          );
+                          const result = await queryPgClient<
+                            Record<string, unknown>
+                          >(pgClient, batch.text, batch.values);
                           totalAffected += result.rowCount ?? 0;
                           if (result.rows) {
                             allPkRows.push(...result.rows);
@@ -200,11 +200,12 @@ export const BulkUpsertPlugin: GraphileConfig.Plugin = {
                           const selectParams = allPkRows.flatMap((pkRow) =>
                             pkColumns.map((col) => pkRow[col])
                           );
-                          const selectResult = await pgClient.query(
+                          const selectResult = await queryPgClient<unknown>(
+                            pgClient,
                             `SELECT * FROM ${compiledFrom} WHERE ${whereClause}`,
                             selectParams
                           );
-                          returning = selectResult.rows || [];
+                          returning = [...selectResult.rows];
                         }
 
                         return {

graphile/graphile-bulk-mutations/src/utils/pg-client.ts

@@ -0,0 +1,10 @@
+import type { PgClient, PgClientResult } from '@dataplan/pg';
+
+/** Execute SQL using @dataplan/pg's native query-config contract. */
+export function queryPgClient<TData>(
+  client: Pick<PgClient, 'query'>,
+  text: string,
+  values: any[]
+): Promise<PgClientResult<TData>> {
+  return client.query<TData>({ text, values });
+}

graphile/graphile-i18n/package.json

@@ -29,6 +29,7 @@
     "url": "https://github.com/constructive-io/constructive/issues"
   },
   "dependencies": {
+    "@constructive-io/express-context": "workspace:^",
     "accept-language-parser": "^1.5.0"
   },
   "peerDependencies": {

graphile/graphile-i18n/src/__tests__/i18n.test.ts

@@ -8,6 +8,7 @@
  * - Fallback to base table values when no translation exists
  */
 
+import { buildPgSettings } from '@constructive-io/express-context';
 import type { GraphQLResponse } from 'graphile-test';
 import { getConnections, seed } from 'graphile-test';
 import { join } from 'path';
@@ -70,7 +71,28 @@ describe('graphile-i18n plugin', () => {
 
     db = connections.db;
     teardown = connections.teardown;
-    query = connections.query;
+    const baseQuery: QueryFn = connections.query;
+    const canonicalSettings = buildPgSettings({
+      api: {
+        apiId: 'i18n-test-api',
+        databaseId: 'i18n-test-database',
+        dbname: 'i18n_test',
+        anonRole: 'postgres',
+        roleName: 'postgres',
+        schema: ['i18n_test'],
+      },
+      token: { user_id: 'i18n-test-user' },
+      requestId: 'i18n-test-request',
+    });
+    query = (document, variables, commit, reqOptions = {}) =>
+      baseQuery(document, variables, commit, {
+        ...reqOptions,
+        pgSettings: {
+          ...canonicalSettings,
+          ...((reqOptions.pgSettings as Record<string, string> | undefined) ??
+            {}),
+        },
+      });
   });
 
   afterAll(async () => {

graphile/graphile-i18n/src/__tests__/pg-query.test.ts

@@ -0,0 +1,79 @@
+import { buildPgSettings } from '@constructive-io/express-context';
+
+import { queryI18nWithContext } from '../pg-query';
+
+const pgSettings = buildPgSettings({
+  api: {
+    apiId: 'api-1',
+    databaseId: 'database-1',
+    dbname: 'testdb',
+    anonRole: 'anonymous_runtime',
+    roleName: 'authenticated_runtime',
+    schema: ['i18n_test'],
+  },
+  token: { user_id: 'user-1' },
+  requestId: 'request-1',
+});
+
+describe('queryI18nWithContext', () => {
+  it('passes the complete settings unchanged and uses the native query contract', async () => {
+    const original = { ...pgSettings };
+    const query = jest.fn(async () => ({
+      rows: [{ lang_code: 'en', title: 'Hello' }],
+    }));
+    const withPgClient = jest.fn(async (settings, callback) =>
+      callback({ query })
+    );
+
+    await expect(
+      queryI18nWithContext(
+        withPgClient,
+        pgSettings,
+        1,
+        'SELECT translation WHERE id = $1 AND lang = ANY($2)',
+        [1, ['en']]
+      )
+    ).resolves.toEqual({ lang_code: 'en', title: 'Hello' });
+
+    expect(withPgClient).toHaveBeenCalledWith(pgSettings, expect.any(Function));
+    expect(withPgClient.mock.calls[0][0]).not.toBeNull();
+    expect(query).toHaveBeenCalledWith({
+      text: 'SELECT translation WHERE id = $1 AND lang = ANY($2)',
+      values: [1, ['en']],
+    });
+    expect(pgSettings).toEqual(original);
+  });
+
+  it.each([
+    [
+      'missing withPgClient',
+      undefined,
+      pgSettings,
+      1,
+      'I18N_PG_CLIENT_CONTEXT_UNAVAILABLE',
+    ],
+    ['missing pgSettings', jest.fn(), undefined, 1, 'i18n pgSettings'],
+    [
+      'incomplete pgSettings',
+      jest.fn(),
+      { role: 'anonymous_runtime' },
+      1,
+      'i18n pgSettings',
+    ],
+  ])(
+    'fails closed for %s',
+    async (_label, withPgClient, settings, id, message) => {
+      await expect(
+        queryI18nWithContext(withPgClient, settings, id, 'SELECT 1', [])
+      ).rejects.toThrow(message);
+    }
+  );
+
+  it('preserves the base-row fallback when the parent id is unavailable', async () => {
+    const withPgClient = jest.fn();
+    await expect(
+      queryI18nWithContext(withPgClient, pgSettings, null, 'SELECT 1', [])
+    ).resolves.toBeNull();
+    expect(withPgClient).not.toHaveBeenCalled();
+  });
+});