Add JobRequest and properly implement event tx

Author: nbsp

agents/package.json

@@ -25,7 +25,10 @@
     "typescript": "^5.0.0"
   },
   "dependencies": {
+    "@livekit/protocol": "^1.12.0",
     "commander": "^12.0.0",
-    "@livekit/protocol": "^1.12.0"
+    "livekit-server-sdk": "^2.1.2",
+    "pino": "^8.19.0",
+    "pino-pretty": "^11.0.0"
   }
 }

agents/src/job_context.ts

@@ -4,16 +4,24 @@
 
 import { Room, LocalParticipant, RemoteParticipant } from '@livekit/rtc-node';
 import { Job } from '@livekit/protocol';
+import { EventEmitter } from 'events';
 
 export class JobContext {
   #job: Job;
   #room: Room;
   #publisher: RemoteParticipant | undefined;
-
-  constructor(job: Job, room: Room, publisher: RemoteParticipant | undefined = undefined) {
+  tx: EventEmitter;
+
+  constructor(
+    tx: EventEmitter,
+    job: Job,
+    room: Room,
+    publisher: RemoteParticipant | undefined = undefined,
+  ) {
     this.#job = job;
     this.#room = room;
     this.#publisher = publisher;
+    this.tx = tx;
   }
 
   get id(): string {
@@ -36,8 +44,7 @@ export class JobContext {
     return this.#room.localParticipant;
   }
 
-  // TODO(nbsp): aio
-  // async shutdown() {
-  //   await this.closeTx.close();
-  // }
+  async shutdown() {
+    this.tx.removeAllListeners();
+  }
 }

agents/src/job_request.ts

@@ -0,0 +1,120 @@
+import { JobContext } from './job_context';
+import { VideoGrant } from 'livekit-server-sdk';
+import { Job, ParticipantInfo, Room } from '@livekit/protocol';
+import { log } from './log';
+import { EventEmitter } from 'events';
+
+class AnsweredError extends Error {
+  constructor() {
+    super();
+    this.name = 'AnsweredError';
+    this.message = 'request already answered';
+  }
+}
+
+enum AutoDisconnect {
+  ROOM_EMPTY,
+  PUBLISHER_LEFT,
+  NONE,
+}
+
+enum AutoSubscribe {
+  SUBSCRIBE_ALL,
+  SUBSCRIBE_NONE,
+  VIDEO_ONLY,
+  AUDIO_ONLY,
+}
+
+type AgentEntry = (ctx: JobContext) => Promise<void>;
+
+type AcceptData = {
+  entry: AgentEntry;
+  autoSubscribe: AutoSubscribe;
+  autoDisconnect: AutoDisconnect;
+  grants: VideoGrant;
+  name: string;
+  identity: string;
+  metadata: string;
+  assign: EventEmitter;
+};
+
+type AvailRes = {
+  avail: boolean;
+  data: AcceptData | undefined;
+};
+
+export class JobRequest {
+  #job: Job;
+  #answered = false;
+  tx: EventEmitter;
+  logger = log.child({ job: this.job });
+
+  constructor(job: Job, tx: EventEmitter) {
+    this.#job = job;
+    this.tx = tx;
+  }
+
+  get id(): string {
+    return this.#job.id;
+  }
+
+  get job(): Job {
+    return this.#job;
+  }
+
+  get room(): Room | undefined {
+    return this.#job.room;
+  }
+
+  get publisher(): ParticipantInfo | undefined {
+    return this.#job.participant;
+  }
+
+  get answered(): boolean {
+    return this.#answered;
+  }
+
+  async reject() {
+    if (this.#answered) {
+      throw new AnsweredError();
+    }
+    this.#answered = true;
+    this.tx.emit('recv', { avail: false, data: undefined } as AvailRes);
+    this.logger.info('rejected job', this.id);
+  }
+
+  async accept(
+    entry: AgentEntry,
+    autoSubscribe: AutoSubscribe = AutoSubscribe.SUBSCRIBE_ALL,
+    autoDisconnect: AutoDisconnect = AutoDisconnect.ROOM_EMPTY,
+    grants: VideoGrant,
+    name: string = '',
+    identity: string = '',
+    metadata: string = '',
+  ) {
+    if (this.#answered) {
+      throw new AnsweredError();
+    }
+    this.#answered = true;
+
+    const assign = new EventEmitter();
+    assign.on('error', (e) => {
+      throw e;
+    });
+
+    const data: AcceptData = {
+      entry,
+      autoSubscribe,
+      autoDisconnect,
+      grants,
+      name,
+      identity,
+      metadata,
+      assign,
+    };
+
+    this.tx.emit('recv', { avail: true, data } as AvailRes);
+
+    this.logger.info('accepted job', this.id);
+  }
+}

agents/src/log.ts

@@ -0,0 +1,10 @@
+import pino from 'pino';
+
+export const log = pino({
+  transport: {
+    target: 'pino-pretty',
+    options: {
+      colorize: true,
+    },
+  },
+});

bun.lockb

@@ -1,3 +1,4 @@
 {
+  "private": true,
   "workspaces": ["agents"]
 }