schema purge

Author: n1ru4l

packages/services/api/src/index.ts

@@ -18,7 +18,6 @@ export type {
   OrganizationBilling,
   OrganizationInvitation,
 } from './shared/entities';
-export { createTaskRunner } from './modules/shared/lib/task-runner';
 export { minifySchema } from './shared/schema';
 export { HiveError } from './shared/errors';
 export { ProjectType } from './__generated__/types';

packages/services/api/src/modules/shared/lib/task-runner.ts

@@ -16,7 +16,6 @@ import { z } from 'zod';
 import formDataPlugin from '@fastify/formbody';
 import {
   createRegistry,
-  createTaskRunner,
   CryptoProvider,
   LogFn,
   Logger,
@@ -39,7 +38,6 @@ import {
   registerTRPC,
   reportReadiness,
   startMetrics,
-  traceInline,
   TracingInstance,
 } from '@hive/service-common';
 import { createConnectionString, createStorage as createPostgreSQLStorage } from '@hive/storage';
@@ -209,50 +207,6 @@ export async function main() {
     }),
   }) as HivePubSub;
 
-  let dbPurgeTaskRunner: null | ReturnType<typeof createTaskRunner> = null;
-
-  if (!env.hiveServices.commerce) {
-    server.log.debug('Commerce service is disabled. Skip scheduling purge tasks.');
-  } else {
-    server.log.debug(
-      `Commerce service is enabled. Start scheduling purge tasks every ${env.hiveServices.commerce.dateRetentionPurgeIntervalMinutes} minutes.`,
-    );
-    dbPurgeTaskRunner = createTaskRunner({
-      run: traceInline(
-        'Purge Task',
-        {
-          resultAttributes: result => ({
-            'purge.schema.check.count': result.deletedSchemaCheckCount,
-            'purge.sdl.store.count': result.deletedSdlStoreCount,
-            'purge.change.approval.count': result.deletedSchemaChangeApprovalCount,
-            'purge.contract.approval.count': result.deletedContractSchemaChangeApprovalCount,
-          }),
-        },
-        async () => {
-          try {
-            const result = await storage.purgeExpiredSchemaChecks({
-              expiresAt: new Date(),
-            });
-            server.log.debug(
-              'Finished running schema check purge task. (deletedSchemaCheckCount=%s deletedSdlStoreCount=%s)',
-              result.deletedSchemaCheckCount,
-              result.deletedSdlStoreCount,
-            );
-
-            return result;
-          } catch (error) {
-            captureException(error);
-            throw error;
-          }
-        },
-      ),
-      interval: env.hiveServices.commerce.dateRetentionPurgeIntervalMinutes * 60 * 1000,
-      logger: server.log,
-    });
-
-    dbPurgeTaskRunner.start();
-  }
-
   registerShutdown({
     logger: server.log,
     async onShutdown() {
@@ -262,10 +216,6 @@ export async function main() {
       await server.close();
       server.log.info('Stopping Storage handler...');
       await storage.destroy();
-      if (dbPurgeTaskRunner) {
-        server.log.info('Stopping expired schema check purge task...');
-        await dbPurgeTaskRunner.stop();
-      }
       server.log.info('Shutdown complete.');
     },
   });

packages/services/server/src/index.ts

@@ -1,6 +1,7 @@
 import zod from 'zod';
 import { OpenTelemetryConfigurationModel } from '@hive/service-common';
 import { createConnectionString } from '@hive/storage';
+import { RequestBroker } from './lib/webhooks/schema-change-notification';
 
 const isNumberString = (input: unknown) => zod.string().regex(/^\d+$/).safeParse(input).success;
 
@@ -81,6 +82,17 @@ const EmailProviderModel = zod.union([
   SendmailEmailModel,
 ]);
 
+const RequestBrokerModel = zod.union([
+  zod.object({
+    REQUEST_BROKER: emptyString(zod.literal('0').optional()),
+  }),
+  zod.object({
+    REQUEST_BROKER: zod.literal('1'),
+    REQUEST_BROKER_ENDPOINT: zod.string().min(1),
+    REQUEST_BROKER_SIGNATURE: zod.string().min(1),
+  }),
+]);
+
 const PrometheusModel = zod.object({
   PROMETHEUS_METRICS: emptyString(zod.union([zod.literal('0'), zod.literal('1')]).optional()),
   PROMETHEUS_METRICS_LABEL_INSTANCE: emptyString(zod.string().optional()),
@@ -112,6 +124,7 @@ const configs = {
   prometheus: PrometheusModel.safeParse(process.env),
   log: LogModel.safeParse(process.env),
   tracing: OpenTelemetryConfigurationModel.safeParse(process.env),
+  requestBroker: RequestBrokerModel.safeParse(process.env),
 };
 
 const environmentErrors: Array<string> = [];
@@ -142,6 +155,7 @@ const sentry = extractConfig(configs.sentry);
 const prometheus = extractConfig(configs.prometheus);
 const log = extractConfig(configs.log);
 const tracing = extractConfig(configs.tracing);
+const requestBroker = extractConfig(configs.requestBroker);
 
 const emailProviderConfig =
   email.EMAIL_PROVIDER === 'postmark'
@@ -211,4 +225,11 @@ export const env = {
       user: postgres.POSTGRES_USER,
     }),
   },
+  requestBroker:
+    requestBroker.REQUEST_BROKER === '1'
+      ? ({
+          endpoint: requestBroker.REQUEST_BROKER_ENDPOINT,
+          signature: requestBroker.REQUEST_BROKER_SIGNATURE,
+        } satisfies RequestBroker)
+      : null,
 } as const;

packages/services/workflows/src/environment.ts

@@ -15,9 +15,10 @@ const pg = await createPool(env.postgres.connectionString);
 const modules = await Promise.all([
   import('./tasks/audit-log-export.js'),
   import('./tasks/email-verification.js'),
-  import('./tasks/organization-invite.js'),
+  import('./tasks/organization-invitation.js'),
   import('./tasks/organization-ownership-transfer.js'),
   import('./tasks/password-reset.js'),
+  import('./tasks/purge-expired-schema-checks.js'),
   import('./tasks/schema-change-notification.js'),
   import('./tasks/usage-rate-limit-exceeded.js'),
   import('./tasks/usage-rate-limit-warning.js'),
@@ -29,6 +30,7 @@ const context: Context = {
   logger,
   email: createEmailProvider(env.email.provider, env.email.emailFrom),
   pg,
+  requestBroker: env.requestBroker,
 };
 
 function logLevel(level: GraphileLogLevel) {
@@ -52,8 +54,10 @@ let runner: Runner = await run({
   logger: new GraphileLogger(_scope => (level, message, _meta) => {
     logger[logLevel(level)](message);
   }),
-  // TODO: define cron jobs!
-  crontab: ' ',
+  crontab: `
+    # Purge expired schema checks every Saturday at 10:00AM
+    0 10 * * 0 purgeExpiredSchemaChecks
+  `,
   connectionString: env.postgres.connectionString,
   taskList: Object.fromEntries(modules.map(module => module.task(context))),
 });

packages/services/workflows/src/index.ts

@@ -1,19 +1,13 @@
-import { JobHelpers, Task } from 'graphile-worker';
+import type { JobHelpers, Task } from 'graphile-worker';
 import { z } from 'zod';
-import { Logger } from '@graphql-hive/logger';
-import { Context } from './context';
+import type { Logger } from '@graphql-hive/logger';
+import type { Context } from './context';
 
 export type TaskDefinition<TName extends string, TModel> = {
   name: TName;
   schema: z.ZodTypeAny & { _output: TModel };
 };
 
-export function defineTask<TName extends string, TModel>(
-  workflow: TaskDefinition<TName, TModel>,
-): TaskDefinition<TName, TModel> {
-  return workflow;
-}
-
 type TaskImplementationArgs<TPayload> = {
   input: TPayload;
   context: Context;
@@ -25,31 +19,52 @@ export type TaskImplementation<TPayload> = (
   args: TaskImplementationArgs<TPayload>,
 ) => Promise<void>;
 
+/**
+ * Define a task
+ */
+export function defineTask<TName extends string, TModel>(
+  workflow: TaskDefinition<TName, TModel>,
+): TaskDefinition<TName, TModel> {
+  return workflow;
+}
+
+/**
+ * Implement a task.
+ */
 export function implementTask<TPayload>(
   taskDefinition: TaskDefinition<string, TPayload>,
   implementation: TaskImplementation<TPayload>,
 ): (context: Context) => [string, Task] {
+  const schema = z.object({
+    requestId: z.string().optional(),
+    input: taskDefinition.schema,
+  });
+
   return function (context) {
     return [
       taskDefinition.name,
       function (unsafePayload, helpers) {
-        const input = taskDefinition.schema.parse(unsafePayload);
+        const payload = schema.parse(unsafePayload);
         return implementation({
-          input,
+          input: payload.input,
           context,
           helpers,
           logger: context.logger.child({
-            attrs: {
-              'job.id': helpers.job.id,
-              'job.queueId': helpers.job.job_queue_id,
-              'job.attempts': helpers.job.attempts,
-              'job.maxAttempts': helpers.job.max_attempts,
-              'job.priority': helpers.job.priority,
-              'job.taskId': helpers.job.task_id,
-            },
+            'request.id': payload.requestId,
+            'job.id': helpers.job.id,
+            'job.queueId': helpers.job.job_queue_id,
+            'job.attempts': helpers.job.attempts,
+            'job.maxAttempts': helpers.job.max_attempts,
+            'job.priority': helpers.job.priority,
+            'job.taskId': helpers.job.task_id,
           }),
         });
       },
     ];
   };
 }
+
+/**
+ * Schedule a task.
+ */
+export function scheduleTask<TPayload>(taskDefinition: TaskDefinition<string, TPayload>) {}

packages/services/workflows/src/kit.ts

@@ -3,7 +3,7 @@ import { defineTask, implementTask } from '../kit.js';
 import { renderAuditLogsReportEmail } from '../lib/emails/templates/audit-logs-report.js';
 
 export const AuditLogExportTask = defineTask({
-  name: 'audit-log-export',
+  name: 'auditLogExport',
   schema: z.object({
     organizationId: z.string(),
     organizationName: z.string(),

packages/services/workflows/src/tasks/audit-log-export.ts

@@ -4,9 +4,14 @@ import { purgeExpiredSchemaChecks } from '../lib/expired-schema-checks';
 
 export const PurgeExpiredSchemaChecks = defineTask({
   name: 'purgeExpiredSchemaChecks',
-  schema: z.undefined(),
+  schema: z.unknown(),
 });
 
 export const task = implementTask(PurgeExpiredSchemaChecks, async args => {
-  await purgeExpiredSchemaChecks({ pool: args.context.pg, expiresAt: new Date() });
+  args.logger.debug('purging expired schema checks');
+  const statistics = await purgeExpiredSchemaChecks({
+    pool: args.context.pg,
+    expiresAt: new Date(),
+  });
+  args.logger.debug({ statistics }, 'finished purging schema checks');
 });