Task Namespace in realtime-api (#471)

* new playground for task

* add support for contexts

* prettier

* add task namespace

* e2e tests for task

* changesets

* cleanup

* add task types

* type task worker

* remove Proxy usage in createTaskObject

* rm debug in e2e tasks

* solved fixme

* rename to task.received

* drop Job and expose send

* fixes

* update changeset from patch to minor

.changeset/khaki-buses-hear.md

@@ -0,0 +1,7 @@
+---
+'@sw-internal/e2e-realtime-api': minor
+'@sw-internal/playground-realtime-api': minor
+'@signalwire/core': minor
+---
+
+[internal] Add playground and e2e tests for Task namespace

.changeset/thin-oranges-pump.md

@@ -0,0 +1,5 @@
+---
+'@signalwire/realtime-api': minor
+---
+
+Add `Task` namespace

internal/e2e-realtime-api/src/task.test.ts

@@ -0,0 +1,46 @@
+import { Task } from '@signalwire/realtime-api'
+import { createTestRunner } from './utils'
+
+const handler = () => {
+ return new Promise<number>(async (resolve, reject) => {
+ const context = 'task-e2e'
+ const jobPayload = {
+ id: Date.now(),
+ item: 'foo',
+ }
+
+ const client = new Task.Client({
+ host: process.env.RELAY_HOST as string,
+ project: process.env.RELAY_PROJECT as string,
+ token: process.env.RELAY_TOKEN as string,
+ contexts: [context],
+ })
+
+ client.on('task.received', (payload) => {
+ if (payload.id === jobPayload.id && payload.item === 'foo') {
+ return resolve(0)
+ }
+ console.error('Invalid payload on `task.received`', payload)
+ return reject(4)
+ })
+
+ await Task.send({
+ host: process.env.RELAY_HOST as string,
+ project: process.env.RELAY_PROJECT as string,
+ token: process.env.RELAY_TOKEN as string,
+ context,
+ message: jobPayload,
+ })
+ })
+}
+
+async function main() {
+ const runner = createTestRunner({
+ name: 'Task E2E',
+ testHandler: handler,
+ })
+
+ await runner.run()
+}
+
+main()

internal/playground-realtime-api/src/task/index.ts

@@ -0,0 +1,28 @@
+import { Task } from '@signalwire/realtime-api'
+
+const client = new Task.Client({
+ host: process.env.HOST || 'relay.swire.io',
+ project: process.env.PROJECT as string,
+ token: process.env.TOKEN as string,
+
+ contexts: ['office'],
+
+ debug: {
+ logWsTraffic: true,
+ },
+})
+
+client.on('task.received', (payload) => {
+ console.log('Task Received', payload)
+})
+
+setTimeout(async () => {
+ console.log('Sending to the client..')
+ await Task.send({
+ host: process.env.HOST || 'relay.swire.io',
+ project: process.env.PROJECT as string,
+ token: process.env.TOKEN as string,
+ context: 'office',
+ message: { yo: ['bro', 1, true] },
+ })
+}, 2000)

packages/core/src/BaseSession.ts

@@ -240,6 +240,9 @@ export class BaseSession {
  if (this._relayProtocolIsValid()) {
  params.protocol = this.relayProtocol
  }
+ if (this.options.contexts?.length) {
+ params.contexts = this.options.contexts
+ }
  this._rpcConnectResult = await this.execute(RPCConnect(params))
  }
 

packages/core/src/index.ts

@@ -65,6 +65,7 @@ export type {
  CustomSaga,
  PubSubChannel,
  MapToPubSubShape,
+ SDKActions,
 } from './redux/interfaces'
 export * as actions from './redux/actions'
 export * as sagaHelpers from './redux/utils/sagaHelpers'
@@ -77,4 +78,4 @@ export type { RoomSessionRecording, RoomSessionPlayback } from './rooms'
 export const selectors = {
  ...sessionSelectors,
 }
-export { ChatMember, ChatMessage } from './chat'
+export { ChatMember, ChatMessage } from './chat'

packages/core/src/redux/features/pubSub/pubSubSaga.ts

@@ -6,10 +6,7 @@ import {
  getLogger,
 } from '../../../utils'
 import type { EventEmitter } from '../../../utils/EventEmitter'
-import type {
- PubSubChannel,
- PubSubAction,
-} from '../../interfaces'
+import type { PubSubChannel, PubSubAction } from '../../interfaces'
 import { findNamespaceInPayload } from '../shared/namespace'
 
 type PubSubSagaParams = {

packages/core/src/redux/interfaces.ts

@@ -12,10 +12,11 @@ import type {
  VideoAPIEventParams,
  InternalVideoAPIEvent,
  ChatAction,
+ TaskAction,
  SwEventParams,
 } from '../types'
 import { SDKRunSaga } from '.'
-import { MulticastChannel } from '@redux-saga/core'
+import { END, MulticastChannel } from '@redux-saga/core'
 
 interface SWComponent {
  id: string
@@ -106,6 +107,9 @@ export type PubSubAction =
  payload: Error | undefined
  }
  | ChatAction
+ | TaskAction
 
 export type PubSubChannel = MulticastChannel<PubSubAction>
 export type SwEventChannel = MulticastChannel<MapToPubSubShape<SwEventParams>>
+
+export type SDKActions = MapToPubSubShape<SwEventParams> | END

packages/core/src/types/index.ts

@@ -3,6 +3,7 @@ import type { VideoAPIEventParams, InternalVideoEventNames } from './video'
 import type { SessionEvents, JSONRPCRequest } from '../utils/interfaces'
 import type { CantinaEvent } from './cantina'
 import type { ChatEvent } from './chat'
+import type { TaskEvent } from './task'
 
 export interface SwEvent {
  event_channel: string
@@ -28,7 +29,7 @@ export interface EmitterContract<
  ): EmitterContract<EventTypes>
 
  removeAllListeners<T extends EventEmitter.EventNames<EventTypes>>(
- event?: T,
+ event?: T
  ): EmitterContract<EventTypes>
 }
 
@@ -115,6 +116,7 @@ export type SwEventParams =
  | WebRTCMessageParams
  | CantinaEvent
  | ChatEvent
+ | TaskEvent
 
 // prettier-ignore
 export type PubSubChannelEvents =
@@ -125,3 +127,4 @@ export * from './video'
 export * from './utils'
 export * from './cantina'
 export * from './chat'
+export * from './task'

packages/core/src/types/task.ts

@@ -0,0 +1,45 @@
+import type { OnlyStateProperties, OnlyFunctionProperties } from '..'
+
+export type TaskReceivedEventName = 'task.received'
+
+export type TaskEventNames = TaskReceivedEventName
+
+export interface TaskContract {}
+
+export type TaskEntity = OnlyStateProperties<TaskContract>
+export type TaskMethods = Omit<
+ OnlyFunctionProperties<TaskContract>,
+ 'subscribe' | 'unsubscribe' | 'updateToken'
+>
+
+/**
+ * ==========
+ * ==========
+ * Server-Side Events
+ * ==========
+ * ==========
+ */
+
+/**
+ * 'queuing.relay.tasks'
+ */
+export interface TaskInboundEvent {
+ event_type: 'queuing.relay.tasks'
+ context: string
+ message: Record<string, unknown>
+ timestamp: number
+ space_id: string
+ project_id: string
+}
+
+export type TaskEvent = TaskInboundEvent
+
+/**
+ * TODO: update MapToPubSubShape in another PR
+ * not used MapToPubSubShape because queuing.relay.tasks
+ * has a different shape
+ */
+export type TaskAction = {
+ type: TaskReceivedEventName
+ payload: TaskInboundEvent
+}

packages/core/src/utils/constants.ts

@@ -22,13 +22,15 @@ export const SYNTHETIC_EVENT_PREFIX = '__synthetic__'
 export const PRODUCT_PREFIX_VIDEO = 'video'
 export const PRODUCT_PREFIX_CANTINA = 'cantina-manager'
 export const PRODUCT_PREFIX_CHAT = 'chat'
+export const PRODUCT_PREFIX_TASK = 'tasking'
 
 export const GLOBAL_VIDEO_EVENTS = ['room.started', 'room.ended'] as const
 
 export const PRODUCT_PREFIXES = [
  PRODUCT_PREFIX_VIDEO,
  PRODUCT_PREFIX_CANTINA,
  PRODUCT_PREFIX_CHAT,
+ PRODUCT_PREFIX_TASK,
 ] as const
 
 /**

packages/core/src/utils/interfaces.ts

@@ -84,6 +84,8 @@ export interface SessionOptions {
  project?: string
  /** SignalWire project token, e.g. `PT9e5660c101cd140a1c93a0197640a369cf5f16975a0079c9` */
  token: string
+ /** Relay contexts, e.g. 'home' or 'office' */
+ contexts?: string[]
  // From `LogLevelDesc` of loglevel to simplify our docs
  /** logging level */
  logLevel?: 'trace' | 'debug' | 'info' | 'warn' | 'error' | 'silent'

packages/core/src/utils/toInternalAction.ts

@@ -1,12 +1,23 @@
 import { MapToPubSubShape } from '../redux/interfaces'
 
 export const toInternalAction = <
- T extends { event_type: unknown; params: unknown }
+ T extends { event_type: unknown; params?: unknown }
 >(
  event: T
 ) => {
  const { event_type, params } = event
 
+ /**
+ * queuing.relay.tasks has a slightly different shape:
+ * no nested "params" so we return the whole event.
+ */
+ if (event_type === 'queuing.relay.tasks') {
+ return {
+ type: event_type,
+ payload: event,
+ } as MapToPubSubShape<T>
+ }
+
  return {
  type: event_type,
  payload: params,

packages/realtime-api/src/index.ts

@@ -76,3 +76,6 @@ export * as Chat from './chat/Chat'
 
 /** @ignore */
 export * from './configure'
+
+/** @ignore */
+export * as Task from './task/Task'

packages/realtime-api/src/task/Task.ts

@@ -0,0 +1,42 @@
+import type { ConsumerContract, BaseComponentOptions } from '@signalwire/core'
+import { connect, BaseConsumer } from '@signalwire/core'
+import type { RealTimeTaskApiEvents } from '../types'
+import { RealtimeClient } from '../client/index'
+import { taskWorker } from './workers'
+
+export interface Task extends ConsumerContract<RealTimeTaskApiEvents> {
+ /** @internal */
+ _session: RealtimeClient
+}
+
+/** @internal */
+class TaskAPI extends BaseConsumer<RealTimeTaskApiEvents> {
+ constructor(options: BaseComponentOptions<RealTimeTaskApiEvents>) {
+ super(options)
+
+ this.setWorker('taskWorker', {
+ worker: taskWorker,
+ })
+ this.attachWorkers()
+ this._attachListeners('')
+ }
+}
+
+/** @internal */
+export const createTaskObject = (
+ params: BaseComponentOptions<RealTimeTaskApiEvents>
+): Task => {
+ const task = connect<RealTimeTaskApiEvents, TaskAPI, Task>({
+ store: params.store,
+ Component: TaskAPI,
+ componentListeners: {
+ errors: 'onError',
+ responses: 'onSuccess',
+ },
+ })(params)
+
+ return task
+}
+
+export * from './TaskClient'
+export * from './send'

packages/realtime-api/src/task/TaskClient.docs.ts

@@ -0,0 +1,12 @@
+import type { Task } from './Task'
+
+export interface TaskClientDocs extends Task {
+ new (opts: {
+ /** SignalWire project id, e.g. `a10d8a9f-2166-4e82-56ff-118bc3a4840f` */
+ project: string
+ /** SignalWire project token, e.g. `PT9e5660c101cd140a1c93a0197640a369cf5f16975a0079c9` */
+ token: string
+ /** SignalWire contexts, e.g. 'home', 'office'... */
+ contexts: string[]
+ }): this
+}

packages/realtime-api/src/task/TaskClient.ts

@@ -0,0 +1,58 @@
+import type { AssertSameType, UserOptions } from '@signalwire/core'
+import { setupClient, clientConnect } from '../client/index'
+import { TaskClientDocs } from './TaskClient.docs'
+import type { Task } from './Task'
+import { createTaskObject } from './Task'
+
+interface TaskClientMain extends Task {
+ new (opts: TaskClientOptions): this
+}
+
+interface TaskClient extends AssertSameType<TaskClientMain, TaskClientDocs> {}
+
+/** @ignore */
+export interface TaskClientOptions
+ extends Omit<UserOptions, '_onRefreshToken'> {
+ contexts: string[]
+}
+
+/** @ignore */
+const TaskClient = function (options?: TaskClientOptions) {
+ const { client, store, emitter } = setupClient(options)
+
+ const task = createTaskObject({
+ store,
+ emitter,
+ })
+
+ const taskOn: TaskClient['on'] = (...args) => {
+ clientConnect(client)
+
+ return task.on(...args)
+ }
+ const taskOnce: TaskClient['once'] = (...args) => {
+ clientConnect(client)
+
+ return task.once(...args)
+ }
+
+ const interceptors = {
+ on: taskOn,
+ once: taskOnce,
+ _session: client,
+ } as const
+
+ return new Proxy<Omit<TaskClient, 'new'>>(task, {
+ get(target, prop, receiver) {
+ if (prop in interceptors) {
+ // @ts-expect-error
+ return interceptors[prop]
+ }
+
+ return Reflect.get(target, prop, receiver)
+ },
+ })
+ // For consistency with other constructors we'll make TS force the use of `new`
+} as unknown as { new (options?: TaskClientOptions): TaskClient }
+
+export { TaskClient as Client }