Merge pull request #4468 from alkem-io/server-4401-ingest-space-feedback

Server 4401 ingest space feedback

.env.docker

@@ -24,6 +24,12 @@ RABBITMQ_HOST=rabbitmq
 RABBITMQ_USER=alkemio-admin
 RABBITMQ_PASSWORD=alkemio!
 RABBITMQ_PORT=5672
+
+
+RABBITMQ_INGEST_SPACE_QUEUE=virtual-contributor-ingest-space
+RABBITMQ_INGEST_SPACE_RESULT_QUEUE=virtual-contributor-ingest-space-result
+RABBITMQ_EVENT_BUS_EXCHANGE=event-bus
+
 ALKEMIO_SERVER_ENDPOINT=http://host.docker.internal:4000/graphql
 KRATOS_API_PUBLIC_ENDPOINT=http://kratos:4433/
 [email protected]
@@ -61,5 +67,4 @@ VECTOR_DB_PORT=8000
 [email protected]
 AUTH_ADMIN_PASSWORD=password
 
-
-LOCAL_STORAGE_PATH=/storage
+LOCAL_STORAGE_PATH=/storage

alkemio.yml

@@ -338,6 +338,12 @@ microservices:
       # RabbitMQ password
       password: ${RABBITMQ_PASSWORD}:alkemio!
 
+    # configuration for the event bus used by the AiServer
+    event_bus:
+      exchange: ${RABBITMQ_EVENT_BUS_EXCHANGE}:event-bus
+      ingest_space_queue: ${RABBITMQ_INGEST_SPACE_QUEUE}:virtual-contributor-ingest-space
+      ingest_space_result_queue: ${RABBITMQ_INGEST_SPACE_RESULT_QUEUE}:virtual-contributor-ingest-space-result
+
 # integrations with 3rd party services
 integrations:
   # Different types of geo information like lat, lan, city, country, etc...

quickstart-services-ai.yml

@@ -439,7 +439,7 @@ services:
       - 'host.docker.internal:host-gateway'
     container_name: alkemio_dev_virtual-contributor-ingest-space
     hostname: virtual-contributor-ingest-space
-    image: alkemio/virtual-contributor-ingest-space:v0.8.1
+    image: alkemio/virtual-contributor-ingest-space:v0.9.0
     platform: linux/x86_64
     restart: always
     volumes:
@@ -453,7 +453,9 @@ services:
       - RABBITMQ_USER
       - RABBITMQ_PASSWORD
       - RABBITMQ_PORT
-      - RABBITMQ_QUEUE=virtual-contributor-ingest-space
+      - RABBITMQ_INGEST_SPACE_QUEUE
+      - RABBITMQ_INGEST_SPACE_RESULT_QUEUE
+      - RABBITMQ_EVENT_BUS_EXCHANGE
       - ENVIRONMENT=dev
       - LOG_LEVEL=debug
       - EMBEDDINGS_DEPLOYMENT_NAME

src/common/constants/providers.ts

@@ -1,5 +1,7 @@
 export const SUBSCRIPTION_DISCUSSION_UPDATED =
   'alkemio-subscriptions-discussion-updated';
+export const SUBSCRIPTION_VIRTUAL_CONTRIBUTOR_UPDATED =
+  'alkemio-subscriptions-vritual-contributor-updated';
 export const SUBSCRIPTION_WHITEBOARD_CONTENT =
   'alkemio-subscriptions-whiteboard-content';
 export const SUBSCRIPTION_WHITEBOARD_SAVED =

src/common/enums/logging.context.ts

@@ -64,6 +64,9 @@ export enum LogContext {
   FILE_INTEGRATION = 'file-integration',
   VIRTUAL_CONTRIBUTOR = 'virtual-contributor',
   AI_SERVER = 'ai-server',
+  AI_SERVER_ADAPTER = 'ai-server-adapter',
   AI_PERSONA_SERVICE = 'ai-persona-service',
   AI_PERSONA_SERVICE_ENGINE = 'ai-persona-service-engine',
+  AI_SERVER_EVENT_BUS = 'ai-server-event-bus',
+  SUBSCRIPTION_PUBLISH = 'subscription-publish',
 }

src/common/enums/messaging.queue.ts

@@ -12,6 +12,7 @@ export enum MessagingQueue {
   EXCALIDRAW_EVENTS = 'alkemio-excalidraw-events',
   //
   SUBSCRIPTION_WHITEBOARD_CONTENT = 'alkemio-subscriptions-whiteboard-content',
+  SUBSCRIPTION_VIRTUAL_CONTRIBUTOR_UPDATED = 'alkemio-subscriptions-virtual-contributor-updated',
   SUBSCRIPTION_WHITEBOARD_SAVED = 'alkemio-subscriptions-whiteboard-saved',
   SUBSCRIPTION_PROFILE_VERIFIED_CREDENTIAL = 'alkemio-subscriptions-profile-verified-credential',
   SUBSCRIPTION_CALLOUT_POST_CREATED = 'alkemio-subscriptions-callout-post-created',

src/common/enums/subscription.type.ts

@@ -8,4 +8,5 @@ export enum SubscriptionType {
   SUBSPACE_CREATED = 'subspaceCreated',
   ACTIVITY_CREATED = 'activityCreated',
   ROOM_EVENTS = 'roomEvents',
+  VIRTUAL_CONTRIBUTOR_UPDATED = 'virtualContributorUpdated',
 }

src/common/enums/virtual.contributor.status.enum.ts

@@ -0,0 +1,10 @@
+import { registerEnumType } from '@nestjs/graphql';
+
+export enum VirtualContributorStatus {
+  INITIALIZING = 'initializing',
+  READY = 'ready',
+}
+
+registerEnumType(VirtualContributorStatus, {
+  name: 'VirtualContributorStatus',
+});

src/core/microservices/microservices.module.ts

@@ -18,6 +18,7 @@ import {
   VIRTUAL_CONTRIBUTOR_ENGINE_EXPERT,
   VIRTUAL_CONTRIBUTOR_ENGINE_GUIDANCE,
   SUBSCRIPTION_WHITEBOARD_SAVED,
+  SUBSCRIPTION_VIRTUAL_CONTRIBUTOR_UPDATED,
 } from '@common/constants/providers';
 import { MessagingQueue } from '@common/enums/messaging.queue';
 import {
@@ -62,6 +63,10 @@ const subscriptionConfig: { provide: string; queueName: MessagingQueue }[] = [
     provide: SUBSCRIPTION_WHITEBOARD_SAVED,
     queueName: MessagingQueue.SUBSCRIPTION_WHITEBOARD_SAVED,
   },
+  {
+    provide: SUBSCRIPTION_VIRTUAL_CONTRIBUTOR_UPDATED,
+    queueName: MessagingQueue.SUBSCRIPTION_VIRTUAL_CONTRIBUTOR_UPDATED,
+  },
 ];
 
 const trackingUUID = randomUUID();

src/domain/communication/room/room.event.resolver.subscription.ts

@@ -34,7 +34,7 @@ export class RoomEventResolverSubscription {
       async resolve(
         this: RoomEventResolverSubscription,
         payload,
-        args,
+        _args,
         context
       ) {
         const agentInfo = context.req?.user;
@@ -74,7 +74,7 @@ export class RoomEventResolverSubscription {
     );
 
     const room = await this.roomService.getRoomOrFail(roomID);
-    await this.authorizationService.grantAccessOrFail(
+    this.authorizationService.grantAccessOrFail(
       agentInfo,
       room.authorization,
       AuthorizationPrivilege.READ,

src/domain/community/virtual-contributor/dto/virtual.contributor.updated.subscription.args.ts

@@ -0,0 +1,11 @@
+import { ArgsType, Field } from '@nestjs/graphql';
+import { UUID_NAMEID } from '@domain/common/scalars';
+
+@ArgsType()
+export class VirtualContributorUpdatedSubscriptionArgs {
+  @Field(() => UUID_NAMEID, {
+    description: 'The Virtual Contributor to receive the events for.',
+    nullable: false,
+  })
+  virtualContributorID!: string;
+}

src/domain/community/virtual-contributor/dto/virtual.contributor.updated.subscription.result.ts

@@ -0,0 +1,13 @@
+import { Field, ObjectType } from '@nestjs/graphql';
+import { IVirtualContributor } from '../virtual.contributor.interface';
+
+@ObjectType('VirtualContributorUpdatedSubscriptionResult', {
+  description: 'The result from a Virtual Contributor update',
+})
+export class VirtualContributorUpdatedSubscriptionResult {
+  @Field(() => IVirtualContributor, {
+    description: 'The Virtual Contributor that was updated',
+    nullable: false,
+  })
+  virtualContributor!: IVirtualContributor;
+}

src/domain/community/virtual-contributor/virtual.contributor.module.ts

@@ -17,6 +17,8 @@ import { AiServerAdapterModule } from '@services/adapters/ai-server-adapter/ai.s
 import { PlatformAuthorizationPolicyModule } from '@platform/authorization/platform.authorization.policy.module';
 import { AccountHostModule } from '@domain/space/account.host/account.host.module';
 import { ContributorModule } from '../contributor/contributor.module';
+import { VirtualContributorResolverSubscriptions } from './virtual.contributor.resolver.subscriptions';
+import { SubscriptionServiceModule } from '@services/subscriptions/subscription-service';
 
 @Module({
   imports: [
@@ -32,13 +34,15 @@ import { ContributorModule } from '../contributor/contributor.module';
     AccountHostModule,
     TypeOrmModule.forFeature([VirtualContributor]),
     PlatformAuthorizationPolicyModule,
+    SubscriptionServiceModule,
   ],
   providers: [
     VirtualContributorService,
     VirtualContributorAuthorizationService,
     VirtualContributorResolverMutations,
     VirtualContributorResolverQueries,
     VirtualContributorResolverFields,
+    VirtualContributorResolverSubscriptions,
   ],
   exports: [VirtualContributorService, VirtualContributorAuthorizationService],
 })

src/domain/community/virtual-contributor/virtual.contributor.resolver.fields.ts

@@ -22,6 +22,7 @@ import { IVirtualContributor } from './virtual.contributor.interface';
 import { IAccount } from '@domain/space/account/account.interface';
 import { IAiPersona } from '../ai-persona';
 import { IContributor } from '../contributor/contributor.interface';
+import { VirtualContributorStatus } from '@common/enums/virtual.contributor.status.enum';
 
 @Resolver(() => IVirtualContributor)
 export class VirtualContributorResolverFields {
@@ -135,4 +136,23 @@ export class VirtualContributorResolverFields {
   ): Promise<IContributor> {
     return await this.virtualContributorService.getProvider(virtualContributor);
   }
+
+  @AuthorizationAgentPrivilege(AuthorizationPrivilege.READ)
+  @ResolveField('status', () => VirtualContributorStatus, {
+    nullable: false,
+    description: 'The status of the virtual contributor',
+  })
+  async status(
+    @Parent() virtualContributor: IVirtualContributor
+  ): Promise<VirtualContributorStatus> {
+    const lastUpdated =
+      await this.virtualContributorService.getBodyOfKnowledgeLastUpdated(
+        virtualContributor
+      );
+
+    if (!!lastUpdated) {
+      return VirtualContributorStatus.READY;
+    }
+    return VirtualContributorStatus.INITIALIZING;
+  }
 }

src/domain/community/virtual-contributor/virtual.contributor.resolver.mutations.ts

@@ -1,9 +1,9 @@
-import { UseGuards } from '@nestjs/common';
+import { Inject, LoggerService, UseGuards } from '@nestjs/common';
 import { Args, Resolver, Mutation, ObjectType } from '@nestjs/graphql';
 import { VirtualContributorService } from './virtual.contributor.service';
 import { CurrentUser, Profiling } from '@src/common/decorators';
 import { GraphqlGuard } from '@core/authorization';
-import { AuthorizationPrivilege } from '@common/enums';
+import { AuthorizationPrivilege, LogContext } from '@common/enums';
 import { AgentInfo } from '@core/authentication.agent.info/agent.info';
 import { AuthorizationService } from '@core/authorization/authorization.service';
 import { IVirtualContributor } from './virtual.contributor.interface';
@@ -12,13 +12,16 @@ import {
   UpdateVirtualContributorInput,
 } from './dto';
 import { RefreshVirtualContributorBodyOfKnowledgeInput } from './dto/virtual.contributor.dto.refresh.body.of.knowlege';
+import { WINSTON_MODULE_NEST_PROVIDER } from 'nest-winston';
 
 @ObjectType('MigrateEmbeddings')
 @Resolver(() => IVirtualContributor)
 export class VirtualContributorResolverMutations {
   constructor(
     private virtualContributorService: VirtualContributorService,
-    private authorizationService: AuthorizationService
+    private authorizationService: AuthorizationService,
+    @Inject(WINSTON_MODULE_NEST_PROVIDER)
+    private readonly logger: LoggerService
   ) {}
 
   @UseGuards(GraphqlGuard)
@@ -80,6 +83,11 @@ export class VirtualContributorResolverMutations {
     @Args('refreshData')
     refreshData: RefreshVirtualContributorBodyOfKnowledgeInput
   ): Promise<boolean> {
+    this.logger.verbose?.(
+      `Refresh body of knowledge mutation invoked for VC ${refreshData.virtualContributorID}`,
+      LogContext.VIRTUAL_CONTRIBUTOR
+    );
+
     const virtual =
       await this.virtualContributorService.getVirtualContributorOrFail(
         refreshData.virtualContributorID,
@@ -89,6 +97,11 @@ export class VirtualContributorResolverMutations {
           },
         }
       );
+    this.logger.verbose?.(
+      `VC ${refreshData.virtualContributorID} found`,
+      LogContext.VIRTUAL_CONTRIBUTOR
+    );
+
     this.authorizationService.grantAccessOrFail(
       agentInfo,
       virtual.authorization,

src/domain/community/virtual-contributor/virtual.contributor.resolver.subscriptions.ts

@@ -0,0 +1,73 @@
+import { Args, Resolver } from '@nestjs/graphql';
+import { CurrentUser, TypedSubscription } from '@src/common/decorators';
+import { VirtualContributorService } from './virtual.contributor.service';
+import { AgentInfo } from '@core/authentication.agent.info/agent.info';
+import { AuthorizationService } from '@core/authorization/authorization.service';
+import { AuthorizationPrivilege, LogContext } from '@common/enums';
+import { Inject, LoggerService, UseGuards } from '@nestjs/common';
+import { GraphqlGuard } from '@core/authorization';
+import { SubscriptionReadService } from '@services/subscriptions/subscription-service';
+import { WINSTON_MODULE_NEST_PROVIDER } from 'nest-winston';
+import { VirtualContributorUpdatedSubscriptionArgs } from './dto/virtual.contributor.updated.subscription.args';
+import { VirtualContributorUpdatedSubscriptionResult } from './dto/virtual.contributor.updated.subscription.result';
+import { VirtualContributorUpdatedSubscriptionPayload } from '@services/subscriptions/subscription-service/dto';
+
+@Resolver()
+export class VirtualContributorResolverSubscriptions {
+  constructor(
+    private virtualContributorService: VirtualContributorService,
+    private authorizationService: AuthorizationService,
+    private subscriptionService: SubscriptionReadService,
+    @Inject(WINSTON_MODULE_NEST_PROVIDER)
+    private readonly logger: LoggerService
+  ) {}
+
+  @UseGuards(GraphqlGuard)
+  @TypedSubscription<
+    VirtualContributorUpdatedSubscriptionPayload,
+    VirtualContributorUpdatedSubscriptionArgs
+  >(() => VirtualContributorUpdatedSubscriptionResult, {
+    description: 'Receive updates on virtual contributors',
+    resolve(this: VirtualContributorResolverSubscriptions, payload) {
+      return payload;
+    },
+    async filter(
+      this: VirtualContributorResolverSubscriptions,
+      payload,
+      variables
+    ) {
+      const isMatch =
+        variables.virtualContributorID === payload.virtualContributor.nameID;
+
+      this.logger.verbose?.(
+        `[Filtering VirtualContribuor updated event id '${payload.eventID}'; VC id ${payload.virtualContributor.nameID}- match=${isMatch}`,
+        LogContext.SUBSCRIPTIONS
+      );
+      return isMatch;
+    },
+  })
+  async virtualContributorUpdated(
+    @CurrentUser() agentInfo: AgentInfo,
+    @Args({ nullable: false })
+    { virtualContributorID }: VirtualContributorUpdatedSubscriptionArgs
+  ) {
+    const vc =
+      await this.virtualContributorService.getVirtualContributorOrFail(
+        virtualContributorID
+      );
+
+    this.authorizationService.grantAccessOrFail(
+      agentInfo,
+      vc.authorization,
+      AuthorizationPrivilege.READ,
+      `subscription to Virtual Contributor updates on: ${vc.id}`
+    );
+
+    this.logger.verbose?.(
+      `Subscribing for updates for VC ${virtualContributorID}`,
+      LogContext.SUBSCRIPTIONS
+    );
+
+    return this.subscriptionService.subscribeToVirtualContributorUpdated();
+  }
+}

src/domain/community/virtual-contributor/virtual.contributor.service.ts

@@ -556,6 +556,13 @@ export class VirtualContributorService {
     return virtualContributorMatchesCount;
   }
 
+  async getBodyOfKnowledgeLastUpdated(virtualContributor: IVirtualContributor) {
+    const aiPersona = await this.getAiPersonaOrFail(virtualContributor);
+    return this.aiServerAdapter.getBodyOfKnowledgeLastUpdated(
+      aiPersona.aiPersonaServiceID
+    );
+  }
+
   //adding this to avoid circular dependency between VirtualContributor, Room, and Invitation
   private async deleteVCInvitations(contributorID: string) {
     const invitations = await this.entityManager.find(Invitation, {

src/domain/space/space/dto/space.dto.ingest.ts

@@ -1,6 +1,6 @@
 import { InputType, Field } from '@nestjs/graphql';
 import { UUID } from '@domain/common/scalars/scalar.uuid';
-import { SpaceIngestionPurpose } from '@services/infrastructure/event-bus/commands';
+import { SpaceIngestionPurpose } from '@services/infrastructure/event-bus/messages';
 
 @InputType()
 export class IngestSpaceInput {

src/migrations/1724333243087-addBoKLastUpdateToAiPersonaService.ts

@@ -0,0 +1,23 @@
+import { MigrationInterface, QueryRunner } from 'typeorm';
+
+export class AddBoKLastUpdateToAiPersonaService1724333243087
+  implements MigrationInterface
+{
+  name = 'AddBoKLastUpdateToAiPersonaService1724333243087';
+
+  public async up(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(
+      `ALTER TABLE \`ai_persona_service\` ADD \`bodyOfKnowledgeLastUpdated\` datetime DEFAULT NULL`
+    );
+
+    await queryRunner.query(
+      `UPDATE \`ai_persona_service\` SET \`bodyOfKnowledgeLastUpdated\`= NOW()`
+    );
+  }
+
+  public async down(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(
+      `ALTER TABLE \`ai_persona_service\` DROP COLUMN \`bodyOfKnowledgeLastUpdated\``
+    );
+  }
+}

src/platform/forum/forum.resolver.subscriptions.ts

@@ -48,6 +48,7 @@ export class ForumResolverSubscriptions {
         payload.discussionID
       );
     },
+
     async filter(
       this: ForumResolverSubscriptions,
       payload: ForumDiscussionUpdated,