feat: queue assistant jobs

chainjet · Dec 16, 2023 · a7e2f49 · a7e2f49
1 parent 03c3cbc
commit a7e2f49
Show file tree

Hide file tree

Showing 5 changed files with 101 additions and 1 deletion.
diff --git a/apps/api/src/chat/chat.module.ts b/apps/api/src/chat/chat.module.ts
@@ -94,6 +94,14 @@ import { OrderService } from './services/order.service'
         maxStalledCount: 3,
       },
     }),
+    BullModule.registerQueue({
+      name: 'assistants',
+      settings: {
+        lockDuration: 60000,
+        stalledInterval: 30000,
+        maxStalledCount: 3,
+      },
+    }),
     CommonModule,
     AuthModule, // required for GraphqlGuard
     UsersModule, // required for GraphqlGuard

diff --git a/apps/api/src/chat/entities/assistant.ts b/apps/api/src/chat/entities/assistant.ts
@@ -18,28 +18,52 @@ export class Assistant extends BaseEntity {
   @prop({ ref: User, required: true })
   readonly owner!: Reference<User>
 
+  @Field()
+  @prop({ required: true })
+  name: string
+
   @Field()
   @prop({ required: true })
   instructions: string
 
+  @Field(() => [String], { nullable: true })
+  @prop({ type: () => [String] })
+  tags?: string[]
+
   @Field({ nullable: true })
+  @prop()
   enabled?: boolean
+
+  @prop()
+  assistantId: string
 }
 
 @InputType()
 export class CreateAssistantInput {
+  @Field()
+  name: string
+
   @Field()
   instructions: string
 
+  @Field(() => [String], { nullable: true })
+  tags?: string[]
+
   @Field({ nullable: true })
   enabled?: boolean
 }
 
 @InputType()
 export class UpdateAssistantInput {
+  @Field({ nullable: true })
+  name: string
+
   @Field({ nullable: true })
   instructions: string
 
+  @Field(() => [String], { nullable: true })
+  tags?: string[]
+
   @Field({ nullable: true })
   enabled?: boolean
 }
diff --git a/apps/api/src/chat/services/assistant.service.ts b/apps/api/src/chat/services/assistant.service.ts
@@ -1,6 +1,12 @@
 import { BaseService } from '@app/common/base/base.service'
+import { RedisPubSubService } from '@app/common/cache/redis-pubsub.service'
+import { InjectQueue } from '@nestjs/bull'
 import { Injectable, Logger } from '@nestjs/common'
+import { DeleteOneOptions, UpdateOneOptions } from '@ptc-org/nestjs-query-core'
 import { ReturnModelType } from '@typegoose/typegoose'
+import { Queue } from 'bull'
+import { UpdateResult } from 'mongodb'
+import { FilterQuery, UpdateQuery } from 'mongoose'
 import { InjectModel } from 'nestjs-typegoose'
 import { Assistant } from '../entities/assistant'
 
@@ -9,8 +15,54 @@ export class AssistantService extends BaseService<Assistant> {
   protected readonly logger = new Logger(AssistantService.name)
   static instance: AssistantService
 
-  constructor(@InjectModel(Assistant) protected readonly model: ReturnModelType<typeof Assistant>) {
+  constructor(
+    @InjectModel(Assistant) protected readonly model: ReturnModelType<typeof Assistant>,
+    @InjectQueue('assistants') private assistantsQueue: Queue,
+    protected redisPubSubService: RedisPubSubService,
+  ) {
     super(model)
     AssistantService.instance = this
   }
+
+  async afterCreateOne(assistant: Assistant) {
+    this.assistantsQueue.add({
+      type: 'created',
+      id: assistant._id,
+    })
+  }
+
+  updateOne(
+    id: string,
+    update: Partial<Assistant>,
+    opts?: UpdateOneOptions<Assistant> | undefined,
+  ): Promise<Assistant> {
+    this.assistantsQueue.add({
+      type: 'updated',
+      id,
+    })
+    return super.updateOne(id, update, opts)
+  }
+
+  updateOneNative(
+    conditions: FilterQuery<new () => Assistant>,
+    query: UpdateQuery<new () => Assistant>,
+  ): Promise<UpdateResult> {
+    this.assistantsQueue.add({
+      type: 'updated',
+      id: conditions._id,
+    })
+    return super.updateOneNative(conditions, query)
+  }
+
+  async deleteOne(id: string, opts?: DeleteOneOptions<Assistant> | undefined): Promise<Assistant> {
+    const assistant = await this.findById(id)
+    if (assistant) {
+      this.assistantsQueue.add({
+        type: 'deleted',
+        id,
+        assistantId: assistant.assistantId,
+      })
+    }
+    return super.deleteOne(id, opts)
+  }
 }
diff --git a/generated/graphql.ts b/generated/graphql.ts
@@ -887,7 +887,9 @@ export interface CreateOneAssistantInput {
 }
 
 export interface CreateAssistantInput {
+    name: string;
     instructions: string;
+    tags?: Nullable<string[]>;
     enabled?: Nullable<boolean>;
 }
 
@@ -897,7 +899,9 @@ export interface UpdateOneAssistantInput {
 }
 
 export interface UpdateAssistantInput {
+    name?: Nullable<string>;
     instructions?: Nullable<string>;
+    tags?: Nullable<string[]>;
     enabled?: Nullable<boolean>;
 }
 
@@ -1399,7 +1403,9 @@ export interface WorkflowRunActionConnection {
 export interface Assistant {
     id: string;
     createdAt: DateTime;
+    name: string;
     instructions: string;
+    tags?: Nullable<string[]>;
     enabled?: Nullable<boolean>;
 }
 
@@ -1421,7 +1427,9 @@ export interface Campaign {
 export interface AssistantDeleteResponse {
     id?: Nullable<string>;
     createdAt?: Nullable<DateTime>;
+    name?: Nullable<string>;
     instructions?: Nullable<string>;
+    tags?: Nullable<string[]>;
     enabled?: Nullable<boolean>;
 }
 

diff --git a/generated/schema.graphql b/generated/schema.graphql
@@ -797,7 +797,9 @@ type WorkflowRunActionConnection {
 type Assistant {
   id: ID!
   createdAt: DateTime!
+  name: String!
   instructions: String!
+  tags: [String!]
   enabled: Boolean
 }
 
@@ -819,7 +821,9 @@ type Campaign {
 type AssistantDeleteResponse {
   id: ID
   createdAt: DateTime
+  name: String
   instructions: String
+  tags: [String!]
   enabled: Boolean
 }
 
@@ -2227,7 +2231,9 @@ input CreateOneAssistantInput {
 }
 
 input CreateAssistantInput {
+  name: String!
   instructions: String!
+  tags: [String!]
   enabled: Boolean
 }
 
@@ -2240,7 +2246,9 @@ input UpdateOneAssistantInput {
 }
 
 input UpdateAssistantInput {
+  name: String
   instructions: String
+  tags: [String!]
   enabled: Boolean
 }