Tom Redman•3w ago

Has anybody solved implementing a trigger for _scheduled_functions changes?

I'm creating a job queue, in which I ctx.schedule jobs, and I'd like to update the resulting jobQueue record when it's complete (my jobQueue objects are different and used differently from the primitive _scheduled_functions). I've tried the convex-helpers/trigger, but it doesn't seem to fire on the _scheduled_functions tables, as I don't update it directly.

import { customCtx, customMutation } from 'convex-helpers/server/customFunctions';
import { Triggers } from 'convex-helpers/server/triggers';

/* eslint-disable no-restricted-imports */
import { internalMutation as rawInternalMutation, mutation as rawMutation } from './_generated/server';

/* eslint-enable no-restricted-imports */
import { DataModel } from './_generated/dataModel';

// start using Triggers, with table types from schema.ts
const triggers = new Triggers<DataModel>();

// register a function to run when a `ctx.db.insert`, `ctx.db.patch`, `ctx.db.replace`, or `ctx.db.delete` changes the "users" table
triggers.register('_scheduled_functions', async (ctx, change) => {
  console.log('Scheduled function changed', change);
});

console.log('Triggers registered');

// create wrappers that replace the built-in `mutation` and `internalMutation`
// the wrappers override `ctx` so that `ctx.db.insert`, `ctx.db.patch`, etc. run registered trigger functions
export const mutation = customMutation(rawMutation, customCtx(triggers.wrapDB));
export const internalMutation = customMutation(rawInternalMutation, customCtx(triggers.wrapDB));

import { customCtx, customMutation } from 'convex-helpers/server/customFunctions';
import { Triggers } from 'convex-helpers/server/triggers';

/* eslint-disable no-restricted-imports */
import { internalMutation as rawInternalMutation, mutation as rawMutation } from './_generated/server';

/* eslint-enable no-restricted-imports */
import { DataModel } from './_generated/dataModel';

// start using Triggers, with table types from schema.ts
const triggers = new Triggers<DataModel>();

// register a function to run when a `ctx.db.insert`, `ctx.db.patch`, `ctx.db.replace`, or `ctx.db.delete` changes the "users" table
triggers.register('_scheduled_functions', async (ctx, change) => {
  console.log('Scheduled function changed', change);
});

console.log('Triggers registered');

// create wrappers that replace the built-in `mutation` and `internalMutation`
// the wrappers override `ctx` so that `ctx.db.insert`, `ctx.db.patch`, etc. run registered trigger functions
export const mutation = customMutation(rawMutation, customCtx(triggers.wrapDB));
export const internalMutation = customMutation(rawInternalMutation, customCtx(triggers.wrapDB));

Here's how I'm using the queue:

try {
      await ctx.runMutation(internal.jobs.jobProcessing.markJobStarted, { jobId });
      const job = await ctx.runQuery(internal.jobs.jobQueue.getJob, { jobId });
      if (!job) throw new Error('Job not found');

      console.log(job);

      switch (job.fnType) {
        case 'action':
          await ctx.scheduler.runAfter(0, job.fnHandle as FunctionHandle<'action'>, job.args);
          break;
        case 'mutation':
          await ctx.scheduler.runAfter(0, job.fnHandle as FunctionHandle<'mutation'>, job.args);
          break;
      }
    } catch (error: any) {
      await ctx.runMutation(internal.jobs.jobProcessing.markJobFailed, {
        error: {
          message: error.message,
          stack: error.stack,
        },
        jobId,
      });
    }

try {
      await ctx.runMutation(internal.jobs.jobProcessing.markJobStarted, { jobId });
      const job = await ctx.runQuery(internal.jobs.jobQueue.getJob, { jobId });
      if (!job) throw new Error('Job not found');

      console.log(job);

      switch (job.fnType) {
        case 'action':
          await ctx.scheduler.runAfter(0, job.fnHandle as FunctionHandle<'action'>, job.args);
          break;
        case 'mutation':
          await ctx.scheduler.runAfter(0, job.fnHandle as FunctionHandle<'mutation'>, job.args);
          break;
      }
    } catch (error: any) {
      await ctx.runMutation(internal.jobs.jobProcessing.markJobFailed, {
        error: {
          message: error.message,
          stack: error.stack,
        },
        jobId,
      });
    }

3 Replies

Convex Bot•3w ago

Thanks for posting in <#1088161997662724167>. Reminder: If you have a Convex Pro account, use the Convex Dashboard to file support tickets. - Provide context: What are you trying to achieve, what is the end-user interaction, what are you seeing? (full error message, command output, etc.) - Use search.convex.dev to search Docs, Stack, and Discord all at once. - Additionally, you can post your questions in the Convex Community's <#1228095053885476985> channel to receive a response from AI. - Avoid tagging staff unless specifically instructed. Thank you!

lee•3w ago

Yeah triggers on scheduled jobs aren't implemented. Triggers as currently implemented only run when your code modifies a table, so even if we did run triggers on ctx.scheduler methods, they wouldn't get run when the job completes, because that's not done by your code. Instead, you could make the scheduled functions themselves have wrappers, like runMutationWrapper from https://github.com/get-convex/workpool/blob/main/src/component/lib.ts

GitHub

workpool/src/component/lib.ts at main · get-convex/workpool

Contribute to get-convex/workpool development by creating an account on GitHub.

Tom RedmanOP•3w ago

Excellent! Thanks @lee 🤗

Has anybody solved implementing a trigger for _scheduled_functions changes?

Did you find this page helpful?