feat: add admin mutations to pause/resume groups (#1183)

Author: pregnantboy

packages/backend/src/graphql/admin/mutations/index.ts

@@ -0,0 +1,9 @@
+import { AdminMutationResolvers } from '@/graphql/__generated__/types.generated'
+
+import pauseGroup from './pause-group'
+import resumeGroup from './resume-group'
+
+export default {
+  pauseGroup,
+  resumeGroup,
+} satisfies AdminMutationResolvers

packages/backend/src/graphql/admin/mutations/pause-group.ts

@@ -0,0 +1,24 @@
+import { actionQueuesByName } from '@/queues/action'
+
+import type { AdminMutationResolvers } from '../../__generated__/types.generated'
+
+const pauseGroup: AdminMutationResolvers['pauseGroup'] = async (
+  _parent,
+  params,
+  _context,
+) => {
+  const {
+    input: { groupId, appKey },
+  } = params
+
+  const queueName = `{app-actions-${appKey}}`
+  const queue = actionQueuesByName[queueName]
+
+  if (!queue) {
+    throw new Error(`Queue ${queueName} not found`)
+  }
+
+  return queue.pauseGroup(groupId)
+}
+
+export default pauseGroup

packages/backend/src/graphql/admin/mutations/resume-group.ts

@@ -0,0 +1,24 @@
+import { actionQueuesByName } from '@/queues/action'
+
+import type { AdminMutationResolvers } from '../../__generated__/types.generated'
+
+const resumeGroup: AdminMutationResolvers['resumeGroup'] = async (
+  _parent,
+  params,
+  _context,
+) => {
+  const {
+    input: { groupId, appKey },
+  } = params
+
+  const queueName = `{app-actions-${appKey}}`
+  const queue = actionQueuesByName[queueName]
+
+  if (!queue) {
+    throw new Error(`Queue ${queueName} not found`)
+  }
+
+  return queue.resumeGroup(groupId)
+}
+
+export default resumeGroup

packages/backend/src/graphql/admin/queries/get-paused-groups.ts

@@ -0,0 +1,24 @@
+import { GroupStatus } from '@taskforcesh/bullmq-pro'
+
+import { actionQueuesByName } from '@/queues/action'
+
+import type { AdminQueryResolvers } from '../../__generated__/types.generated'
+
+const getPausedGroups: AdminQueryResolvers['getPausedGroups'] = async (
+  _parent,
+  params,
+  _context,
+) => {
+  const { appKey } = params
+  const queueName = `{app-actions-${appKey}}`
+  const queue = actionQueuesByName[queueName]
+
+  if (!queue) {
+    throw new Error(`Queue ${queueName} not found`)
+  }
+
+  const pausedGroups = await queue.getGroupsByStatus(GroupStatus.Paused)
+  return pausedGroups.map((group) => group.id)
+}
+
+export default getPausedGroups

packages/backend/src/graphql/admin/queries/index.ts

@@ -2,6 +2,7 @@ import type { AdminQueryResolvers } from '../../__generated__/types.generated'
 
 import getExecutionOwner from './get-execution-owner'
 import getFlowOwner from './get-flow-owner'
+import getPausedGroups from './get-paused-groups'
 import getTableOwner from './get-table-owner'
 import searchUsersByEmail from './search-users-by-email'
 
@@ -10,4 +11,5 @@ export default {
   getFlowOwner,
   getTableOwner,
   searchUsersByEmail,
+  getPausedGroups,
 } satisfies AdminQueryResolvers

packages/backend/src/graphql/mutation-resolvers.ts

@@ -84,4 +84,9 @@ export default {
   deleteUploadedFile,
   generatePresignedUrl,
   ...tilesMutationResolvers,
+
+  // This is a special stub that enables us to group all our admin-related
+  // mutations into a special AdminMutation object; each "mutation" is handled by field
+  // resolvers defined in @/graphql/admin/mutations.
+  admin: () => ({}),
 } satisfies MutationResolvers

packages/backend/src/graphql/resolvers.ts

@@ -1,4 +1,5 @@
 import type { Resolvers } from './__generated__/types.generated'
+import adminMutationResolvers from './admin/mutations'
 import adminQueryResolvers from './admin/queries'
 import customResolvers from './custom-resolvers'
 import mutationResolvers from './mutation-resolvers'
@@ -8,5 +9,6 @@ export default {
   Query: queryResolvers,
   Mutation: mutationResolvers,
   AdminQuery: adminQueryResolvers,
+  AdminMutation: adminMutationResolvers,
   ...customResolvers,
 } satisfies Resolvers

packages/backend/src/graphql/schema.graphql

@@ -60,8 +60,10 @@ type AdminQuery {
   getExecutionOwner(executionId: String!): User
   getTableOwner(tableId: String!): User
   searchUsersByEmail(query: String!): [User!]!
+  getPausedGroups(appKey: String!): [String!]!
 }
 
+
 type Mutation {
   createConnection(input: CreateConnectionInput): Connection
   generateAuthUrl(input: GenerateAuthUrlInput): AuthLink
@@ -117,6 +119,14 @@ type Mutation {
   # Flow transfers
   createFlowTransfer(input: CreateFlowTransferInput!): FlowTransfer!
   updateFlowTransferStatus(input: UpdateFlowTransferStatusInput): FlowTransfer!
+  # Admin mutations
+  admin: AdminMutation!
+}
+
+
+type AdminMutation {
+  pauseGroup(input: PauseGroupInput!): Boolean
+  resumeGroup(input: ResumeGroupInput!): Boolean
 }
 
 """
@@ -986,6 +996,20 @@ enum TemplateTagType {
 
 # End of template types
 
+# Start of admin types
+
+input PauseGroupInput {
+  groupId: String!
+  appKey: String!
+}
+
+input ResumeGroupInput {
+  groupId: String!
+  appKey: String!
+}
+
+# End of admin types
+
 schema {
   query: Query
   mutation: Mutation

packages/backend/src/helpers/authentication.ts

@@ -104,7 +104,9 @@ const authentication = shield(
       loginWithSgid: rateLimitRule({ window: '1s', max: 5 }),
       loginWithSelectedSgid: rateLimitRule({ window: '1s', max: 5 }),
       loginWithSso: rateLimitRule({ window: '1s', max: 5 }),
+      admin: isAdminOperation,
     },
+    AdminMutation: isAdminOperation,
   },
   {
     allowExternalErrors: true,