Time service (#10546)

Signed-off-by: Denis Bykhov <bykhov.denis@gmail.com>

common/scripts/docker.sh

@@ -57,5 +57,6 @@ else
     --to @hcengineering/pod-billing \
     --to @hcengineering/pod-process \
     --to @hcengineering/pod-rating \
-    --to @hcengineering/pod-payment
+    --to @hcengineering/pod-payment \
+    --to @hcengineering/pod-worker
 fi

dev/docker-compose.yaml

@@ -79,7 +79,7 @@ services:
       - --mode dev-container
       - --smp 1
       - --default-log-level=info
-      - --memory 256M
+      - --memory 512M
     container_name: redpanda
     volumes:
       - redpanda:/var/lib/redpanda/data
@@ -594,6 +594,22 @@ services:
       - QUEUE_CONFIG=${QUEUE_CONFIG}
       - QUEUE_REGION=cockroach
     restart: unless-stopped
+  time-machine:
+    image: hardcoreeng/worker
+    extra_hosts:
+      - 'huly.local:host-gateway'
+    depends_on:
+      redpanda:
+        condition: service_started
+      account:
+        condition: service_started
+      cockroach:
+        condition: service_started
+    environment:
+      - DB_URL=${DB_CR_URL}
+      - QUEUE_CONFIG=${QUEUE_CONFIG}
+      - QUEUE_REGION=cockroach
+    restart: unless-stopped
   #  translate:
   #    image: hardcoreeng/translate
   #    extra_hosts:

foundations/server/packages/core/src/queue/types.ts

@@ -19,7 +19,9 @@ export enum QueueTopic {
   CalendarEventCUD = 'calendarEventCUD',
 
   // A topic about process events
-  Process = 'process'
+  Process = 'process',
+
+  TimeMachine = 'timeMachine'
 }
 
 export interface ConsumerHandle {

server-plugins/process/src/types.ts

@@ -1,7 +1,7 @@
 import { Card } from '@hcengineering/card'
+import { CollaboratorClient } from '@hcengineering/collaborator-client'
 import { Doc, MeasureContext, PersonId, Ref, Timestamp, Tx, TxOperations, WorkspaceUuid } from '@hcengineering/core'
 import { Execution, ExecutionError, MethodParams, Trigger, UserResult } from '@hcengineering/process'
-import { CollaboratorClient } from '@hcengineering/collaborator-client'
 
 export type ExecuteFunc = (
   params: MethodParams<Doc>,
@@ -51,3 +51,11 @@ export interface ProcessControl {
 }
 
 export type RollbackFunc = (context: Record<string, any>, control: ProcessControl) => Tx
+
+export interface TimeMachineMessage {
+  type: 'schedule' | 'cancel'
+  id: string
+  targetDate?: Timestamp
+  topic?: string
+  data?: any
+}

services/process/package.json

@@ -68,7 +68,6 @@
     "@hcengineering/account-client": "workspace:^0.7.21",
     "@hcengineering/communication-types": "workspace:^0.7.12",
     "@hcengineering/communication-sdk-types": "workspace:^0.7.12",
-    "@temporalio/client": "1.12.3",
     "dotenv": "^16.4.5"
   }
 }

services/process/src/index.ts

@@ -25,7 +25,6 @@ import { join } from 'path'
 import config from './config'
 import { prepare } from './init'
 import { messageHandler } from './main'
-import { closeTemporal } from './temporal'
 import { SERVICE_NAME } from './utils'
 
 async function main (): Promise<void> {
@@ -63,7 +62,6 @@ async function main (): Promise<void> {
   )
 
   const shutdown = (): void => {
-    void closeTemporal()
     void Promise.all([consumer.close()]).then(() => {
       process.exit()
     })

services/process/src/main.ts

@@ -14,6 +14,8 @@
 //
 
 import cardPlugin, { Card } from '@hcengineering/card'
+import { CreateMessageEvent, MessageEventType } from '@hcengineering/communication-sdk-types'
+import { ActivityProcess, ActivityUpdateType, MessageType } from '@hcengineering/communication-types'
 import core, {
   Doc,
   generateId,
@@ -30,6 +32,7 @@ import core, {
   TxUpdateDoc,
   WorkspaceUuid
 } from '@hcengineering/core'
+import { getPlatformQueue } from '@hcengineering/kafka'
 import { getResource } from '@hcengineering/platform'
 import process, {
   Execution,
@@ -49,22 +52,19 @@ import process, {
   Trigger,
   UserResult
 } from '@hcengineering/process'
+import { QueueTopic } from '@hcengineering/server-core'
 import serverProcess, {
   ExecuteResult,
   MethodImpl,
   ProcessControl,
   ProcessMessage,
+  TimeMachineMessage,
   TriggerImpl
 } from '@hcengineering/server-process'
 import { getContextValue } from '@hcengineering/server-process-resources'
-import { Client as TemporalClient } from '@temporalio/client'
-import config from './config'
-import { isError } from './errors'
-import { getTemporalClient } from './temporal'
-import { getClient, releaseClient } from './utils'
-import { CreateMessageEvent, MessageEventType } from '@hcengineering/communication-sdk-types'
-import { ActivityUpdateType, ActivityProcess, MessageType } from '@hcengineering/communication-types'
 import { createCollaboratorClient } from './collaborator'
+import { isError } from './errors'
+import { getClient, releaseClient, SERVICE_NAME } from './utils'
 
 const activeExecutions = new Set<Ref<Execution>>()
 
@@ -608,9 +608,8 @@ async function updateExecutionTimers (control: ProcessControl, execution: Execut
       trigger: process.trigger.OnTime
     })
     if (transitions.length === 0) return
-    const temporalClient = await getTemporalClient()
     for (const transition of transitions) {
-      await setTimer(control, execution, transition, temporalClient)
+      await setTimer(control, execution, transition)
     }
   } catch (err) {
     control.ctx.error('Error setting next timers:', { error: err, execution: execution._id })
@@ -619,66 +618,60 @@ async function updateExecutionTimers (control: ProcessControl, execution: Execut
 
 async function setNextTimers (control: ProcessControl, execution: Execution): Promise<void> {
   try {
-    const temporalClient = await getTemporalClient()
-    await cleanTimers(execution, temporalClient)
+    await cleanTimers(control, execution)
     const transitions = control.client.getModel().findAllSync(process.class.Transition, {
       from: execution.currentState,
       process: execution.process,
       trigger: process.trigger.OnTime
     })
     for (const transition of transitions) {
-      await setTimer(control, execution, transition, temporalClient)
+      await setTimer(control, execution, transition)
     }
   } catch (err) {
     control.ctx.error('Error setting next timers:', { error: err, execution: execution._id })
   }
 }
 
-async function cleanTimers (execution: Execution, temporalClient: TemporalClient): Promise<void> {
+async function cleanTimers (control: ProcessControl, execution: Execution): Promise<void> {
   try {
-    const res = await temporalClient.workflowService.listWorkflowExecutions({
-      namespace: config.TemporalNamespace,
-      query: `WorkflowType="processTimeWorkflow" AND ExecutionStatus="Running" AND ProcessExecution="${execution._id}"`
-    })
-
-    for (const ex of res.executions) {
-      try {
-        await temporalClient.workflowService.terminateWorkflowExecution({
-          workflowExecution: {
-            workflowId: ex.execution?.workflowId,
-            runId: ex.execution?.runId
-          },
-          reason: 'Outdated'
-        })
-      } catch (err) {
-        console.error('Error terminating workflow execution:', err)
+    const queue = getPlatformQueue(SERVICE_NAME)
+    const producer = queue.getProducer<TimeMachineMessage>(control.ctx, QueueTopic.TimeMachine)
+    await producer.send(control.ctx, control.workspace, [
+      {
+        type: 'cancel',
+        id: `${execution._id}_%`
       }
-    }
+    ])
   } catch (err) {
     console.error('Error cleaning timers:', err)
   }
 }
 
-async function setTimer (
-  control: ProcessControl,
-  execution: Execution,
-  transition: Transition,
-  temporalClient: TemporalClient
-): Promise<void> {
+async function setTimer (control: ProcessControl, execution: Execution, transition: Transition): Promise<void> {
   const filled = await fillParams(transition.triggerParams, execution, control)
   const targetDate: number = filled.value
   if (targetDate === undefined || typeof targetDate !== 'number' || targetDate === 0 || Number.isNaN(targetDate)) return
   try {
-    await temporalClient.workflow.signalWithStart('processTimeWorkflow', {
-      taskQueue: 'process',
-      signal: 'setDate',
-      args: [targetDate, control.workspace, execution._id],
-      signalArgs: [targetDate],
-      workflowId: `${execution._id}_${transition._id}`,
-      searchAttributes: {
-        ProcessExecution: [execution._id]
+    const queue = getPlatformQueue(SERVICE_NAME)
+    const producer = queue.getProducer<TimeMachineMessage>(control.ctx, QueueTopic.TimeMachine)
+
+    const data: ProcessMessage = {
+      account: core.account.System,
+      event: [process.trigger.OnTime],
+      createdOn: Date.now(),
+      context: {},
+      execution: execution._id
+    }
+
+    await producer.send(control.ctx, control.workspace, [
+      {
+        type: 'schedule',
+        id: `${execution._id}_${transition._id}`,
+        targetDate,
+        topic: QueueTopic.Process,
+        data
       }
-    })
+    ])
   } catch (e) {
     console.error('Error setting timer:', e)
   }

services/process/src/temporal.ts

@@ -1,41 +0,0 @@
-//
-// Copyright © 2025 Hardcore Engineering Inc.
-//
-// Licensed under the Eclipse Public License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License. You may
-// obtain a copy of the License at https://www.eclipse.org/legal/epl-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-//
-// See the License for the specific language governing permissions and
-// limitations under the License.
-//
-
-import { Client, Connection } from '@temporalio/client'
-import config from './config'
-
-let temporalConnection: Connection | Promise<Connection> | undefined
-
-export async function closeTemporal (): Promise<void> {
-  if (temporalConnection !== undefined) {
-    if (temporalConnection instanceof Promise) {
-      temporalConnection = await temporalConnection
-    }
-    await temporalConnection.close()
-  }
-}
-
-export async function getTemporalClient (): Promise<Client> {
-  if (temporalConnection === undefined) {
-    temporalConnection = Connection.connect({
-      address: config.TemporalAddress
-    })
-  }
-  const temporalClient = new Client({
-    connection: await temporalConnection,
-    namespace: config.TemporalNamespace
-  })
-  return temporalClient
-}

services/worker/Dockerfile

@@ -1,15 +1,6 @@
-FROM node:20-bullseye-slim
-
-RUN apt-get update \
-    && apt-get install -y ca-certificates \
-    && rm -rf /var/lib/apt/lists/*
-
+FROM hardcoreeng/base-slim:v20250916
 WORKDIR /usr/src/app
 
-ENV NODE_ENV=production
-RUN npm install --ignore-scripts=false --verbose @temporalio/worker @temporalio/workflow --unsafe-perm
-
 COPY bundle/bundle.js ./
-COPY bundle/workflows.js ./
 
 CMD [ "node", "bundle.js" ]

services/worker/README.md

@@ -0,0 +1,44 @@
+# Time Machine Service
+
+The Time Machine service is an autonomous, generic service responsible for handling delayed events (timers). It replaces the previous Temporal implementation with a database-backed polling mechanism and Kafka-based communication.
+
+## How it works
+
+1.  **Commands**: The service consumes commands from the `TimeMachine` Kafka topic.
+2.  **Storage**: Scheduled events are stored in a PostgreSQL table `time_machine.delayed_events`.
+3.  **Polling**: The service periodically polls the database for expired events.
+4.  **Events**: When an event expires, the service sends the stored `data` to the specified `topic` via Kafka and removes the record from its database.
+
+## Kafka Interactions
+
+### Consumed (Incoming)
+**Topic**: `TimeMachine` (`timeMachine`)
+**Message Type**: `TimeMachineMessage`
+
+| Type | Description |
+| :--- | :--- |
+| `schedule` | Schedules a new timer or updates an existing one. Requires `id`, `targetDate`, `topic`, and `data`. |
+| `cancel` | Removes scheduled timers. The `id` supports pattern matching via `ILIKE` (e.g., `prefix_%`). |
+
+### Produced (Outgoing)
+**Topic**: Dynamic (specified in `schedule` command)
+**Message Type**: Arbitrary JSON (stored in `data`)
+
+When a timer expires, the service relays the exact `data` payload to the target `topic`.
+
+## Environment Variables
+
+| Variable | Default | Description |
+| :--- | :--- | :--- |
+| `DB_URL` | `postgres://localhost:5432/huly` | Connection string for the PostgreSQL database. |
+| `POLL_INTERVAL` | `5000` | Polling interval for expired events in milliseconds. |
+| `QUEUE_CONFIG` | - | Kafka bootstrap servers configuration. |
+| `QUEUE_REGION` | `cockroach` | Platform region configuration. |
+
+## Database Schema
+
+The service automatically initializes its own schema if it doesn't exist:
+- **Schema**: `time_machine`
+- **Table**: `delayed_events`
+- **Columns**: `id` (text), `workspace` (uuid), `target_date` (int8), `topic` (text), `data` (jsonb).
+- **Primary Key**: `(id, workspace)`

services/worker/esbuild.js

@@ -1,26 +0,0 @@
-const esbuild = require('esbuild')
-const fs = require('fs');
-
-fs.mkdirSync('bundle', { recursive: true });
-
-void esbuild.build({
-    entryPoints: ['src/index.ts'],
-    bundle: true,
-    minify: true,
-    keepNames: true,
-    loader: { ".node": "file" },
-    platform: 'node',
-    outfile: 'bundle/bundle.js',
-    external: ['@temporalio/*']
-})
-
-void esbuild.build({
-    entryPoints: ['src/workflows.ts'],
-    bundle: true,
-    minify: true,
-    keepNames: true,
-    loader: { ".node": "file" },
-    platform: 'node',
-    outfile: 'bundle/workflows.js',
-    external: ['@temporalio/*']
-})

services/worker/package.json

@@ -14,7 +14,7 @@
     "build": "compile",
     "build:watch": "compile",
     "test": "jest --passWithNoTests --silent",
-    "bundle": "node esbuild.js",
+    "bundle": "node ../../common/scripts/esbuild.js",
     "_phase:bundle": "rushx bundle",
     "_phase:docker-build": "rushx docker:build",
     "_phase:docker-staging": "rushx docker:staging",
@@ -54,11 +54,11 @@
     "@hcengineering/server-process": "workspace:^0.7.0"
   },
   "dependencies": {
-    "@temporalio/worker": "1.12.3",
-    "@temporalio/workflow": "1.12.3",
     "@hcengineering/kafka": "workspace:^0.7.18",
     "@hcengineering/process": "workspace:^0.7.0",
     "@hcengineering/core": "workspace:^0.7.24",
-    "@hcengineering/server-core": "workspace:^0.7.18"
+    "@hcengineering/server-core": "workspace:^0.7.18",
+    "postgres": "^3.4.7",
+    "dotenv": "^16.4.5"
   }
 }

services/worker/src/activities.ts

@@ -1,25 +1,18 @@
-import core, { MeasureMetricsContext, type Ref, type WorkspaceUuid } from '@hcengineering/core'
+import { MeasureMetricsContext, type WorkspaceUuid } from '@hcengineering/core'
 import { getPlatformQueue } from '@hcengineering/kafka'
-import process, { type Execution } from '@hcengineering/process'
-import { QueueTopic } from '@hcengineering/server-core'
-import type { ProcessMessage } from '@hcengineering/server-process'
 
-export async function SendTimeEvent (ws: WorkspaceUuid, _execution: Ref<Execution>): Promise<void> {
-  const SERVICE_NAME = 'worker'
+export async function SendTimeEvent (
+  ctx: MeasureMetricsContext,
+  ws: WorkspaceUuid,
+  topic: string,
+  data: any
+): Promise<void> {
+  const SERVICE_NAME = 'time-machine'
   const queue = getPlatformQueue(SERVICE_NAME)
-  const ctx = new MeasureMetricsContext(SERVICE_NAME, {})
 
-  const producer = queue.getProducer<ProcessMessage>(ctx, QueueTopic.Process)
+  const producer = queue.getProducer<any>(ctx, topic as any)
 
-  await producer.send(ctx, ws, [
-    {
-      account: core.account.System,
-      event: [process.trigger.OnTime],
-      createdOn: Date.now(),
-      context: {},
-      execution: _execution
-    }
-  ])
+  await producer.send(ctx, ws, [data])
 }
 
 export default {

services/worker/src/config.ts

@@ -0,0 +1,33 @@
+//
+// Copyright © 2025 Hardcore Engineering Inc.
+//
+// Licensed under the Eclipse Public License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License. You may
+// obtain a copy of the License at https://www.eclipse.org/legal/epl-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+import { config as dotenvConfig } from 'dotenv'
+
+dotenvConfig()
+
+export interface Config {
+  DbUrl: string
+  PollInterval: number
+  QueueRegion: string
+  QueueConfig: string
+}
+
+const config: Config = {
+  DbUrl: process.env.DB_URL ?? 'postgres://localhost:5432/huly',
+  PollInterval: process.env.POLL_INTERVAL != null ? Number(process.env.POLL_INTERVAL) : 20000,
+  QueueRegion: process.env.QUEUE_REGION ?? 'cockroach',
+  QueueConfig: process.env.QUEUE_CONFIG ?? ''
+}
+
+export default config

services/worker/src/db.ts

@@ -0,0 +1,107 @@
+//
+// Copyright © 2025 Hardcore Engineering Inc.
+//
+// Licensed under the Eclipse Public License, Version 2.0 (the 'License');
+// you may not use this file except in compliance with the License. You may
+// obtain a copy of the License at https://www.eclipse.org/legal/epl-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an 'AS IS' BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+
+import postgres from 'postgres'
+import { WorkspaceUuid } from '@hcengineering/core'
+
+export interface DelayedEventRecord {
+  id: string
+  workspace: WorkspaceUuid
+  target_date: number
+  topic: string
+  data: any
+}
+
+const delayedEventsTable = 'time_machine.delayed_events'
+
+export class TimeMachineDB {
+  constructor (private readonly client: postgres.Sql) {}
+
+  static async init (dbUrl: string): Promise<TimeMachineDB> {
+    const client = postgres(dbUrl, {
+      connection: {
+        application_name: 'time-machine'
+      }
+    })
+
+    const sql = `
+        CREATE SCHEMA IF NOT EXISTS time_machine;
+        
+        CREATE TABLE IF NOT EXISTS ${delayedEventsTable} (
+          id TEXT NOT NULL,
+          workspace UUID NOT NULL,
+          target_date INT8 NOT NULL,
+          topic TEXT NOT NULL,
+          data JSONB NOT NULL,
+          PRIMARY KEY (id, workspace)
+        );
+
+        CREATE INDEX IF NOT EXISTS idx_delayed_events_target_date ON ${delayedEventsTable} (target_date);
+    `
+
+    await client.unsafe(sql)
+    return new TimeMachineDB(client)
+  }
+
+  async upsertEvent (record: DelayedEventRecord): Promise<void> {
+    await this.client`
+      INSERT INTO time_machine.delayed_events (id, workspace, target_date, topic, data)
+      VALUES (${record.id}, ${record.workspace}, ${record.target_date}, ${record.topic}, ${record.data})
+      ON CONFLICT (id, workspace) DO UPDATE 
+      SET target_date = EXCLUDED.target_date,
+          topic = EXCLUDED.topic,
+          data = EXCLUDED.data
+    `
+  }
+
+  async removeEvents (workspace: WorkspaceUuid, idPattern: string): Promise<void> {
+    await this.client`
+      DELETE FROM time_machine.delayed_events 
+      WHERE workspace = ${workspace} AND id ILIKE ${idPattern}
+    `
+  }
+
+  async getExpiredEvents (): Promise<DelayedEventRecord[]> {
+    const now = Date.now()
+    const res = await this.client`
+      SELECT id, workspace, target_date, topic, data 
+      FROM time_machine.delayed_events 
+      WHERE target_date <= ${now}
+    `
+    return res.map((r: any) => ({
+      id: r.id,
+      workspace: r.workspace,
+      target_date: Number(r.target_date),
+      topic: r.topic,
+      data: r.data
+    }))
+  }
+
+  async deleteEvents (events: DelayedEventRecord[]): Promise<void> {
+    if (events.length === 0) return
+    await this.client.begin(async (sql: postgres.Sql) => {
+      for (const event of events) {
+        await sql`
+          DELETE FROM time_machine.delayed_events 
+          WHERE id = ${event.id} AND workspace = ${event.workspace}
+        `
+      }
+    })
+  }
+
+  async close (): Promise<void> {
+    await this.client.end()
+  }
+}

services/worker/src/worker.ts

@@ -13,24 +13,57 @@
 // limitations under the License.
 //
 
-import { NativeConnection, Worker } from '@temporalio/worker'
-import * as activities from './activities'
+import { MeasureMetricsContext } from '@hcengineering/core'
+import { getPlatformQueue } from '@hcengineering/kafka'
+import { QueueTopic } from '@hcengineering/server-core'
+import { TimeMachineMessage } from '@hcengineering/server-process'
+import { TimeMachineDB } from './db'
+import { SendTimeEvent } from './activities'
+import config from './config'
 
 export async function runWorker (): Promise<void> {
-  const connection = await NativeConnection.connect({
-    address: process.env.TEMPORAL_ADDRESS ?? 'localhost:7233'
-  })
-  try {
-    const worker = await Worker.create({
-      connection,
-      workflowsPath: require.resolve('./workflows'),
-      activities,
-      namespace: process.env.TEMPORAL_NAMESPACE ?? 'huly',
-      taskQueue: 'process'
-    })
+  const SERVICE_NAME = 'time-machine'
+  const db = await TimeMachineDB.init(config.DbUrl)
 
-    await worker.run()
-  } finally {
-    await connection.close()
+  const ctx = new MeasureMetricsContext(SERVICE_NAME, {})
+  const queue = getPlatformQueue(SERVICE_NAME)
+
+  // 1. Kafka Consumer for commands
+  queue.createConsumer<TimeMachineMessage>(ctx, QueueTopic.TimeMachine, SERVICE_NAME, async (ctx, msg) => {
+    const { type, id, targetDate, topic, data } = msg.value
+    if (type === 'schedule' && targetDate != null && topic != null && data !== undefined) {
+      await db.upsertEvent({
+        id,
+        workspace: msg.workspace,
+        target_date: targetDate,
+        topic,
+        data
+      })
+    } else if (type === 'cancel') {
+      await db.removeEvents(msg.workspace, id)
+    }
+  })
+
+  // 2. Polling loop for expired events
+  const poll = async (): Promise<void> => {
+    try {
+      const expiredEvents = await db.getExpiredEvents()
+      if (expiredEvents.length > 0) {
+        for (const event of expiredEvents) {
+          await SendTimeEvent(ctx, event.workspace, event.topic, event.data)
+        }
+        await db.deleteEvents(expiredEvents)
+      }
+    } catch (err) {
+      ctx.error('Error in Time Machine polling loop:')
+    } finally {
+      setTimeout(() => {
+        void poll()
+      }, config.PollInterval)
+    }
   }
+
+  void poll()
+
+  ctx.info('Time Machine worker started')
 }

services/worker/src/workflows.ts

@@ -1,62 +0,0 @@
-//
-// Copyright © 2025 Hardcore Engineering Inc.
-//
-// Licensed under the Eclipse Public License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License. You may
-// obtain a copy of the License at https://www.eclipse.org/legal/epl-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-//
-// See the License for the specific language governing permissions and
-// limitations under the License.
-//
-
-import type { Ref, WorkspaceUuid } from '@hcengineering/core'
-import type { Execution } from '@hcengineering/process'
-import { defineSignal, proxyActivities, setHandler, sleep } from '@temporalio/workflow'
-import activities from './activities'
-
-const { SendTimeEvent } = proxyActivities<typeof activities>({
-  startToCloseTimeout: '10 minute'
-})
-
-const setDate = defineSignal<[number]>('setDate')
-
-export async function processTimeWorkflow (
-  _targetDate: number,
-  ws: WorkspaceUuid,
-  _execution: Ref<Execution>
-): Promise<void> {
-  let targetDate: number = _targetDate
-  let currentPromise: Promise<void> | undefined
-
-  while (true) {
-    const when = new Date(targetDate).getTime()
-    const delay = when - Date.now()
-
-    if (delay > 0) {
-      currentPromise = sleep(delay)
-      try {
-        await Promise.race([
-          currentPromise,
-          new Promise((resolve, reject) => {
-            setHandler(setDate, (newDate) => {
-              targetDate = newDate
-              reject(new Error('Date updated'))
-            })
-          })
-        ])
-      } catch (error: any) {
-        if (error instanceof Error && error.message === 'Date updated') {
-          continue
-        }
-        throw error
-      }
-    }
-
-    await SendTimeEvent(ws, _execution)
-    break
-  }
-}

tests/docker-compose.yaml

@@ -11,6 +11,21 @@ services:
     ports:
       - 1081:1080
     restart: unless-stopped
+  time-machine:
+    image: hardcoreeng/worker
+    pull_policy: never
+    depends_on:
+      redpanda:
+        condition: service_started
+      account:
+        condition: service_started
+      postgres:
+        condition: service_healthy
+    environment:
+      - DB_URL=postgres://postgres:postgres@postgres:5433/postgres
+      - QUEUE_CONFIG=${QUEUE_CONFIG}
+      - QUEUE_REGION=
+    restart: unless-stopped
   mongodb:
     image: 'mongo:7-jammy'
     command: mongod --port 27018