mirror of
https://github.com/hcengineering/platform.git
synced 2026-08-17 18:05:42 +02:00
feat: Add ability to schedule notifications (#10789)
* feat: Add ability to schedule notifications Signed-off-by: Artem Savchenko <armisav@gmail.com> * Clean up Signed-off-by: Artem Savchenko <armisav@gmail.com> * Clean up Signed-off-by: Artem Savchenko <armisav@gmail.com> * Add docker file Signed-off-by: Artem Savchenko <armisav@gmail.com> * Rename pod Signed-off-by: Artem Savchenko <armisav@gmail.com> * Add debug logging Signed-off-by: Artem Savchenko <armisav@gmail.com> * Reminder fixes Signed-off-by: Artem Savchenko <armisav@gmail.com> * Fix reminders Signed-off-by: Artem Savchenko <armisav@gmail.com> * Support reminders for all events Signed-off-by: Artem Savchenko <armisav@gmail.com> * Clean up Signed-off-by: Artem Savchenko <armisav@gmail.com> * Support for project todo Signed-off-by: Artem Savchenko <armisav@gmail.com> * Fix mismatched dependency Signed-off-by: Artem Savchenko <armisav@gmail.com> * Use base event class Signed-off-by: Artem Savchenko <armisav@gmail.com> --------- Signed-off-by: Artem Savchenko <armisav@gmail.com>
This commit is contained in:
@@ -114,3 +114,4 @@ dev/tool/history.json
|
||||
.tmp
|
||||
ws-tests/docker-compose.override.yml
|
||||
.cursor/*
|
||||
CLAUDE.md
|
||||
@@ -99,6 +99,12 @@ For detailed information about the platform architecture, services, and their in
|
||||
- [Docker](https://docs.docker.com/get-docker/)
|
||||
- [Docker Compose](https://docs.docker.com/compose/install/)
|
||||
|
||||
If you use `nvm`, run this after entering the repo to align your shell with the repository Node version:
|
||||
|
||||
```bash
|
||||
nvm use
|
||||
```
|
||||
|
||||
## Verification
|
||||
|
||||
To verify the installation, perform the following checks in your terminal:
|
||||
|
||||
Generated
+109
@@ -40218,6 +40218,115 @@ importers:
|
||||
specifier: ^5.9.3
|
||||
version: 5.9.3
|
||||
|
||||
../../services/notification/pod-events-processor:
|
||||
dependencies:
|
||||
'@hcengineering/account-client':
|
||||
specifier: workspace:^0.7.25
|
||||
version: link:../../../foundations/core/packages/account-client
|
||||
'@hcengineering/analytics':
|
||||
specifier: workspace:^0.7.19
|
||||
version: link:../../../foundations/core/packages/analytics
|
||||
'@hcengineering/analytics-service':
|
||||
specifier: workspace:^0.7.19
|
||||
version: link:../../../foundations/core/packages/analytics-service
|
||||
'@hcengineering/api-client':
|
||||
specifier: workspace:^0.7.25
|
||||
version: link:../../../foundations/core/packages/api-client
|
||||
'@hcengineering/calendar':
|
||||
specifier: workspace:^0.7.0
|
||||
version: link:../../../plugins/calendar
|
||||
'@hcengineering/contact':
|
||||
specifier: workspace:^0.7.0
|
||||
version: link:../../../plugins/contact
|
||||
'@hcengineering/core':
|
||||
specifier: workspace:^0.7.26
|
||||
version: link:../../../foundations/core/packages/core
|
||||
'@hcengineering/kafka':
|
||||
specifier: workspace:^0.7.18
|
||||
version: link:../../../foundations/server/packages/kafka
|
||||
'@hcengineering/notification':
|
||||
specifier: workspace:^0.7.0
|
||||
version: link:../../../plugins/notification
|
||||
'@hcengineering/platform':
|
||||
specifier: workspace:^0.7.20
|
||||
version: link:../../../foundations/core/packages/platform
|
||||
'@hcengineering/server-client':
|
||||
specifier: workspace:^0.7.16
|
||||
version: link:../../../foundations/server/packages/client
|
||||
'@hcengineering/server-core':
|
||||
specifier: workspace:^0.7.19
|
||||
version: link:../../../foundations/server/packages/core
|
||||
'@hcengineering/server-token':
|
||||
specifier: workspace:^0.7.18
|
||||
version: link:../../../foundations/core/packages/token
|
||||
'@hcengineering/text-core':
|
||||
specifier: workspace:^0.7.19
|
||||
version: link:../../../foundations/core/packages/text-core
|
||||
'@hcengineering/time':
|
||||
specifier: workspace:^0.7.0
|
||||
version: link:../../../plugins/time
|
||||
dotenv:
|
||||
specifier: ^16.4.5
|
||||
version: 16.6.1
|
||||
devDependencies:
|
||||
'@hcengineering/platform-rig':
|
||||
specifier: workspace:^0.7.21
|
||||
version: link:../../../foundations/utils/packages/platform-rig
|
||||
'@tsconfig/node16':
|
||||
specifier: ^1.0.4
|
||||
version: 1.0.4
|
||||
'@types/jest':
|
||||
specifier: ^29.5.5
|
||||
version: 29.5.14
|
||||
'@types/node':
|
||||
specifier: ^22.18.1
|
||||
version: 22.19.0
|
||||
'@typescript-eslint/eslint-plugin':
|
||||
specifier: ^6.21.0
|
||||
version: 6.21.0(@typescript-eslint/parser@6.21.0(eslint@8.57.1)(typescript@5.9.3))(eslint@8.57.1)(typescript@5.9.3)
|
||||
'@typescript-eslint/parser':
|
||||
specifier: ^6.21.0
|
||||
version: 6.21.0(eslint@8.57.1)(typescript@5.9.3)
|
||||
cross-env:
|
||||
specifier: ~7.0.3
|
||||
version: 7.0.3
|
||||
esbuild:
|
||||
specifier: ^0.25.10
|
||||
version: 0.25.12
|
||||
eslint:
|
||||
specifier: ^8.54.0
|
||||
version: 8.57.1
|
||||
eslint-config-standard-with-typescript:
|
||||
specifier: ^40.0.0
|
||||
version: 40.0.0(@typescript-eslint/eslint-plugin@6.21.0(@typescript-eslint/parser@6.21.0(eslint@8.57.1)(typescript@5.9.3))(eslint@8.57.1)(typescript@5.9.3))(eslint-plugin-import@2.32.0(eslint@8.57.1))(eslint-plugin-n@15.7.0(eslint@8.57.1))(eslint-plugin-promise@6.6.0(eslint@8.57.1))(eslint@8.57.1)(typescript@5.9.3)
|
||||
eslint-plugin-import:
|
||||
specifier: ^2.26.0
|
||||
version: 2.32.0(eslint@8.57.1)
|
||||
eslint-plugin-n:
|
||||
specifier: ^15.4.0
|
||||
version: 15.7.0(eslint@8.57.1)
|
||||
eslint-plugin-node:
|
||||
specifier: ^11.1.0
|
||||
version: 11.1.0(eslint@8.57.1)
|
||||
eslint-plugin-promise:
|
||||
specifier: ^6.1.1
|
||||
version: 6.6.0(eslint@8.57.1)
|
||||
jest:
|
||||
specifier: ^29.7.0
|
||||
version: 29.7.0(@types/node@22.19.0)(ts-node@10.9.2(@types/node@22.19.0)(typescript@5.9.3))
|
||||
prettier:
|
||||
specifier: ^3.6.2
|
||||
version: 3.6.2
|
||||
ts-jest:
|
||||
specifier: ^29.1.1
|
||||
version: 29.4.5(@babel/core@7.28.5)(@jest/transform@29.7.0)(@jest/types@30.2.0)(babel-jest@29.7.0(@babel/core@7.28.5))(esbuild@0.25.12)(jest-util@30.2.0)(jest@29.7.0(@types/node@22.19.0)(ts-node@10.9.2(@types/node@22.19.0)(typescript@5.9.3)))(typescript@5.9.3)
|
||||
ts-node:
|
||||
specifier: ^10.9.2
|
||||
version: 10.9.2(@types/node@22.19.0)(typescript@5.9.3)
|
||||
typescript:
|
||||
specifier: ^5.9.3
|
||||
version: 5.9.3
|
||||
|
||||
../../services/notification/pod-notification:
|
||||
dependencies:
|
||||
'@hcengineering/analytics':
|
||||
|
||||
@@ -58,5 +58,6 @@ else
|
||||
--to @hcengineering/pod-process \
|
||||
--to @hcengineering/pod-rating \
|
||||
--to @hcengineering/pod-payment \
|
||||
--to @hcengineering/pod-worker
|
||||
--to @hcengineering/pod-worker \
|
||||
--to @hcengineering/pod-events-processor
|
||||
fi
|
||||
|
||||
@@ -595,6 +595,24 @@ services:
|
||||
- QUEUE_CONFIG=${QUEUE_CONFIG}
|
||||
- QUEUE_REGION=cockroach
|
||||
restart: unless-stopped
|
||||
events-processor:
|
||||
image: hardcoreeng/events-processor
|
||||
extra_hosts:
|
||||
- 'huly.local:host-gateway'
|
||||
depends_on:
|
||||
redpanda:
|
||||
condition: service_started
|
||||
account:
|
||||
condition: service_started
|
||||
environment:
|
||||
- SERVICE_ID=events-processor
|
||||
- LOG_LEVEL=debug
|
||||
- SECRET=secret
|
||||
- ACCOUNTS_URL=http://huly.local:3000
|
||||
- QUEUE_CONFIG=${QUEUE_CONFIG}
|
||||
- QUEUE_REGION=cockroach
|
||||
- OTEL_EXPORTER_OTLP_ENDPOINT=http://jaeger:4318/v1/traces
|
||||
restart: unless-stopped
|
||||
# translate:
|
||||
# image: hardcoreeng/translate
|
||||
# extra_hosts:
|
||||
|
||||
@@ -16,7 +16,7 @@ export class SplitLogger implements MeasureLogger {
|
||||
const rootDir = this.opts.root ?? 'logs'
|
||||
|
||||
this.logger = winston.createLogger({
|
||||
level: 'info',
|
||||
level: process.env.LOG_LEVEL === 'debug' ? 'debug' : 'info',
|
||||
exitOnError: false
|
||||
})
|
||||
const errorPrinter = ({ message, stack, ...rest }: Error): object => ({
|
||||
@@ -99,6 +99,13 @@ export class SplitLogger implements MeasureLogger {
|
||||
this.logger.warn({ message, ...obj })
|
||||
}
|
||||
|
||||
debug (message: string, obj?: Record<string, any>): void {
|
||||
if (this.opts.parent !== undefined) {
|
||||
this.opts.parent.debug({ message, ...obj })
|
||||
}
|
||||
this.logger.debug({ message, ...obj })
|
||||
}
|
||||
|
||||
logOperation (operation: string, time: number, params: ParamsType): void {
|
||||
this.logger.info(operation, { time, ...params })
|
||||
}
|
||||
|
||||
@@ -22,6 +22,7 @@ describe('telemetry', () => {
|
||||
info: jest.fn(),
|
||||
error: jest.fn(),
|
||||
warn: jest.fn(),
|
||||
debug: jest.fn(),
|
||||
close: jest.fn(async () => {}),
|
||||
logOperation: jest.fn()
|
||||
}
|
||||
|
||||
@@ -11,6 +11,7 @@ import {
|
||||
updateMeasure,
|
||||
type FullParamsType,
|
||||
type MeasureLogger,
|
||||
type MeasureLogLevel,
|
||||
type Metrics,
|
||||
type ParamsType,
|
||||
type WithOptions
|
||||
@@ -97,7 +98,8 @@ export class OpenTelemetryMetricsContext implements MeasureContext {
|
||||
readonly logParams?: ParamsType,
|
||||
|
||||
readonly otlpLogger?: Logger,
|
||||
readonly meter?: MetricsContext
|
||||
readonly meter?: MetricsContext,
|
||||
readonly logLevel: MeasureLogLevel = 'info'
|
||||
) {
|
||||
this.name = name
|
||||
this.params = params
|
||||
@@ -133,6 +135,7 @@ export class OpenTelemetryMetricsContext implements MeasureContext {
|
||||
logger?: MeasureLogger
|
||||
span?: WithOptions['span'] // By default true
|
||||
meta?: Record<string, string | number | boolean>
|
||||
logLevel?: MeasureLogLevel
|
||||
}
|
||||
): MeasureContext {
|
||||
let _span: Span | undefined
|
||||
@@ -170,7 +173,8 @@ export class OpenTelemetryMetricsContext implements MeasureContext {
|
||||
this,
|
||||
this.logParams,
|
||||
this.otlpLogger,
|
||||
this.meter
|
||||
this.meter,
|
||||
opt?.logLevel ?? this.logLevel
|
||||
)
|
||||
result.id = this.id
|
||||
result.contextData = this.contextData
|
||||
@@ -309,6 +313,23 @@ export class OpenTelemetryMetricsContext implements MeasureContext {
|
||||
this.logger.warn(message, { ...this.params, ...args, ...(this.logParams ?? {}) })
|
||||
}
|
||||
|
||||
debug (message: string, args?: Record<string, any>): void {
|
||||
if (this.logLevel !== 'debug') return
|
||||
if (this.otlpLogger !== undefined) {
|
||||
this.otlpLogger.emit({
|
||||
severityNumber: SeverityNumber.DEBUG,
|
||||
severityText: 'debug',
|
||||
context: this.context,
|
||||
body: message,
|
||||
attributes: {
|
||||
'service.name': sdkServiceName,
|
||||
...(args ?? {})
|
||||
}
|
||||
})
|
||||
}
|
||||
this.logger.debug(message, { ...this.params, ...args, ...(this.logParams ?? {}) })
|
||||
}
|
||||
|
||||
end (): void {
|
||||
this.done()
|
||||
}
|
||||
@@ -530,7 +551,8 @@ export function createOpenTelemetryMetricsContext (
|
||||
): MeasureContext {
|
||||
if (!initOpenTelemetrySDK(name, version ?? '')) {
|
||||
console.warn('OTEL_EXPORTER_OTLP_TRACES_ENDPOINT is not set, OpenTelemetry metrics will not be sent')
|
||||
return new MeasureMetricsContext(name, params, fullParams, metrics, logger)
|
||||
const rootLogLevel: MeasureLogLevel = process.env.LOG_LEVEL === 'debug' ? 'debug' : 'info'
|
||||
return new MeasureMetricsContext(name, params, fullParams, metrics, logger, undefined, undefined, rootLogLevel)
|
||||
}
|
||||
|
||||
// Traces
|
||||
@@ -542,6 +564,8 @@ export function createOpenTelemetryMetricsContext (
|
||||
|
||||
const meter = otelMetrics.getMeter(name, version)
|
||||
|
||||
const rootLogLevel: MeasureLogLevel = process.env.LOG_LEVEL === 'debug' ? 'debug' : 'info'
|
||||
|
||||
const ctx = new OpenTelemetryMetricsContext(
|
||||
name,
|
||||
tracer,
|
||||
@@ -554,7 +578,8 @@ export function createOpenTelemetryMetricsContext (
|
||||
undefined,
|
||||
undefined,
|
||||
otlpLogger,
|
||||
new MetricsContext(meter)
|
||||
new MetricsContext(meter),
|
||||
rootLogLevel
|
||||
)
|
||||
return ctx
|
||||
}
|
||||
|
||||
@@ -21,6 +21,7 @@ describe('context', () => {
|
||||
expect(typeof logger.info).toBe('function')
|
||||
expect(typeof logger.error).toBe('function')
|
||||
expect(typeof logger.warn).toBe('function')
|
||||
expect(typeof logger.debug).toBe('function')
|
||||
expect(typeof logger.close).toBe('function')
|
||||
})
|
||||
|
||||
@@ -54,6 +55,16 @@ describe('context', () => {
|
||||
consoleSpy.mockRestore()
|
||||
})
|
||||
|
||||
it('should log debug messages', () => {
|
||||
const consoleSpy = jest.spyOn(console, 'debug').mockImplementation()
|
||||
const logger = consoleLogger({ service: 'test' })
|
||||
|
||||
logger.debug('Debug message', { detail: 'x' })
|
||||
|
||||
expect(consoleSpy).toHaveBeenCalled()
|
||||
consoleSpy.mockRestore()
|
||||
})
|
||||
|
||||
it('should handle errors in params', () => {
|
||||
const consoleSpy = jest.spyOn(console, 'error').mockImplementation()
|
||||
const logger = consoleLogger({})
|
||||
@@ -145,6 +156,7 @@ describe('context', () => {
|
||||
info: jest.fn(),
|
||||
error: jest.fn(),
|
||||
warn: jest.fn(),
|
||||
debug: jest.fn(),
|
||||
close: jest.fn(async () => {}),
|
||||
logOperation: jest.fn()
|
||||
}
|
||||
@@ -163,6 +175,7 @@ describe('context', () => {
|
||||
info: jest.fn(),
|
||||
error: jest.fn(),
|
||||
warn: jest.fn(),
|
||||
debug: jest.fn(),
|
||||
close: jest.fn(async () => {}),
|
||||
logOperation: jest.fn()
|
||||
}
|
||||
@@ -181,6 +194,7 @@ describe('context', () => {
|
||||
info: jest.fn(),
|
||||
error: jest.fn(),
|
||||
warn: jest.fn(),
|
||||
debug: jest.fn(),
|
||||
close: jest.fn(async () => {}),
|
||||
logOperation: jest.fn()
|
||||
}
|
||||
@@ -194,6 +208,47 @@ describe('context', () => {
|
||||
)
|
||||
})
|
||||
|
||||
it('should not log debug when log level is info', () => {
|
||||
const mockLogger: MeasureLogger = {
|
||||
info: jest.fn(),
|
||||
error: jest.fn(),
|
||||
warn: jest.fn(),
|
||||
debug: jest.fn(),
|
||||
close: jest.fn(async () => {}),
|
||||
logOperation: jest.fn()
|
||||
}
|
||||
|
||||
const ctx = new MeasureMetricsContext('test', { op: 'test' }, {}, newMetrics(), mockLogger)
|
||||
ctx.debug('Skip', { k: 1 })
|
||||
|
||||
expect(mockLogger.debug).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('should log debug when log level is debug', () => {
|
||||
const mockLogger: MeasureLogger = {
|
||||
info: jest.fn(),
|
||||
error: jest.fn(),
|
||||
warn: jest.fn(),
|
||||
debug: jest.fn(),
|
||||
close: jest.fn(async () => {}),
|
||||
logOperation: jest.fn()
|
||||
}
|
||||
|
||||
const ctx = new MeasureMetricsContext(
|
||||
'test',
|
||||
{ op: 'test' },
|
||||
{},
|
||||
newMetrics(),
|
||||
mockLogger,
|
||||
undefined,
|
||||
undefined,
|
||||
'debug'
|
||||
)
|
||||
ctx.debug('D', { k: 1 })
|
||||
|
||||
expect(mockLogger.debug).toHaveBeenCalledWith('D', expect.objectContaining({ k: 1, op: 'test' }))
|
||||
})
|
||||
|
||||
it('should get params', () => {
|
||||
const ctx = new MeasureMetricsContext('test', { op: 'test', method: 'GET' }, {}, newMetrics(), logger)
|
||||
const params = ctx.getParams()
|
||||
@@ -440,6 +495,7 @@ describe('context', () => {
|
||||
info: jest.fn(),
|
||||
error: jest.fn(),
|
||||
warn: jest.fn(),
|
||||
debug: jest.fn(),
|
||||
close: jest.fn(async () => {}),
|
||||
logOperation: jest.fn()
|
||||
}
|
||||
|
||||
@@ -6,6 +6,7 @@ import {
|
||||
type FullParamsType,
|
||||
type MeasureContext,
|
||||
type MeasureLogger,
|
||||
type MeasureLogLevel,
|
||||
type Metrics,
|
||||
type ParamsType,
|
||||
type OperationLog,
|
||||
@@ -42,6 +43,14 @@ export const consoleLogger = (logParams: Record<string, any>): MeasureLogger =>
|
||||
warn: (msg, args) => {
|
||||
console.warn(msg, ...Object.entries(args ?? {}).map((it) => `${it[0]}=${JSON.stringify(replacer(it[1]))}`))
|
||||
},
|
||||
debug: (msg, args) => {
|
||||
console.debug(
|
||||
msg,
|
||||
...Object.entries({ ...(args ?? {}), ...(logParams ?? {}) }).map(
|
||||
(it) => `${it[0]}=${JSON.stringify(replacer(it[1]))}`
|
||||
)
|
||||
)
|
||||
},
|
||||
close: async () => {},
|
||||
logOperation: (operation, time, params) => {}
|
||||
})
|
||||
@@ -62,6 +71,8 @@ export class MeasureMetricsContext implements MeasureContext {
|
||||
metrics: Metrics
|
||||
id?: string
|
||||
|
||||
private readonly logLevel: MeasureLogLevel
|
||||
|
||||
st = platformNow()
|
||||
contextData: object = {}
|
||||
private done (value?: number, override?: boolean): void {
|
||||
@@ -75,12 +86,14 @@ export class MeasureMetricsContext implements MeasureContext {
|
||||
metrics: Metrics = newMetrics(),
|
||||
logger?: MeasureLogger,
|
||||
readonly parent?: MeasureContext,
|
||||
readonly logParams?: ParamsType
|
||||
readonly logParams?: ParamsType,
|
||||
logLevel: MeasureLogLevel = 'info'
|
||||
) {
|
||||
this.name = name
|
||||
this.params = params
|
||||
this.fullParams = fullParams
|
||||
this.metrics = metrics
|
||||
this.logLevel = logLevel
|
||||
this.metrics.namedParams = this.metrics.namedParams ?? {}
|
||||
for (const [k, v] of Object.entries(params)) {
|
||||
if (this.metrics.namedParams[k] !== v) {
|
||||
@@ -94,7 +107,16 @@ export class MeasureMetricsContext implements MeasureContext {
|
||||
}
|
||||
|
||||
measure (name: string, value: number, override?: boolean): void {
|
||||
const c = new MeasureMetricsContext('#' + name, {}, {}, childMetrics(this.metrics, ['#' + name]), this.logger, this)
|
||||
const c = new MeasureMetricsContext(
|
||||
'#' + name,
|
||||
{},
|
||||
{},
|
||||
childMetrics(this.metrics, ['#' + name]),
|
||||
this.logger,
|
||||
this,
|
||||
undefined,
|
||||
this.logLevel
|
||||
)
|
||||
c.contextData = this.contextData
|
||||
c.done(value, override)
|
||||
}
|
||||
@@ -106,6 +128,8 @@ export class MeasureMetricsContext implements MeasureContext {
|
||||
fullParams?: FullParamsType
|
||||
logger?: MeasureLogger
|
||||
span?: WithOptions['span'] // By default true
|
||||
meta?: Record<string, string | number | boolean>
|
||||
logLevel?: MeasureLogLevel
|
||||
}
|
||||
): MeasureContext {
|
||||
const result = new MeasureMetricsContext(
|
||||
@@ -115,7 +139,8 @@ export class MeasureMetricsContext implements MeasureContext {
|
||||
childMetrics(this.metrics, [name]),
|
||||
opt?.logger ?? this.logger,
|
||||
this,
|
||||
this.logParams
|
||||
this.logParams,
|
||||
opt?.logLevel ?? this.logLevel
|
||||
)
|
||||
result.id = this.id
|
||||
result.contextData = this.contextData
|
||||
@@ -187,6 +212,11 @@ export class MeasureMetricsContext implements MeasureContext {
|
||||
this.logger.warn(message, { ...this.params, ...args, ...(this.logParams ?? {}) })
|
||||
}
|
||||
|
||||
debug (message: string, args?: Record<string, any>): void {
|
||||
if (this.logLevel !== 'debug') return
|
||||
this.logger.debug(message, { ...this.params, ...args, ...(this.logParams ?? {}) })
|
||||
}
|
||||
|
||||
end (): void {
|
||||
this.done()
|
||||
}
|
||||
@@ -202,8 +232,11 @@ export class NoMetricsContext implements MeasureContext {
|
||||
|
||||
contextData: object = {}
|
||||
|
||||
constructor (logger?: MeasureLogger) {
|
||||
private readonly logLevel: MeasureLogLevel
|
||||
|
||||
constructor (logger?: MeasureLogger, logLevel: MeasureLogLevel = 'info') {
|
||||
this.logger = logger ?? consoleLogger({})
|
||||
this.logLevel = logLevel
|
||||
}
|
||||
|
||||
measure (name: string, value: number, override?: boolean): void {}
|
||||
@@ -211,10 +244,15 @@ export class NoMetricsContext implements MeasureContext {
|
||||
newChild (
|
||||
name: string,
|
||||
params: ParamsType,
|
||||
fullParams?: FullParamsType | (() => FullParamsType),
|
||||
logger?: MeasureLogger
|
||||
opt?: {
|
||||
fullParams?: FullParamsType | (() => FullParamsType)
|
||||
logger?: MeasureLogger
|
||||
span?: WithOptions['span']
|
||||
meta?: Record<string, string | number | boolean>
|
||||
logLevel?: MeasureLogLevel
|
||||
}
|
||||
): MeasureContext {
|
||||
const result = new NoMetricsContext(logger ?? this.logger)
|
||||
const result = new NoMetricsContext(opt?.logger ?? this.logger, opt?.logLevel ?? this.logLevel)
|
||||
result.id = this.id
|
||||
result.contextData = this.contextData
|
||||
return result
|
||||
@@ -226,7 +264,7 @@ export class NoMetricsContext implements MeasureContext {
|
||||
op: (ctx: MeasureContext) => T | Promise<T>,
|
||||
fullParams?: ParamsType | (() => FullParamsType)
|
||||
): Promise<T> {
|
||||
const r = op(this.newChild(name, params, fullParams, this.logger))
|
||||
const r = op(this.newChild(name, params, { fullParams, logger: this.logger }))
|
||||
return r instanceof Promise ? r : Promise.resolve(r)
|
||||
}
|
||||
|
||||
@@ -240,7 +278,7 @@ export class NoMetricsContext implements MeasureContext {
|
||||
op: (ctx: MeasureContext) => T,
|
||||
fullParams?: ParamsType | (() => FullParamsType)
|
||||
): T {
|
||||
const c = this.newChild(name, params, fullParams, this.logger)
|
||||
const c = this.newChild(name, params, { fullParams, logger: this.logger })
|
||||
return op(c)
|
||||
}
|
||||
|
||||
@@ -250,7 +288,7 @@ export class NoMetricsContext implements MeasureContext {
|
||||
op: (ctx: MeasureContext) => T | Promise<T>,
|
||||
fullParams?: ParamsType
|
||||
): Promise<T> {
|
||||
const r = op(this.newChild(name, params, fullParams, this.logger))
|
||||
const r = op(this.newChild(name, params, { fullParams, logger: this.logger }))
|
||||
return r instanceof Promise ? r : Promise.resolve(r)
|
||||
}
|
||||
|
||||
@@ -266,6 +304,11 @@ export class NoMetricsContext implements MeasureContext {
|
||||
this.logger.warn(message, { ...args })
|
||||
}
|
||||
|
||||
debug (message: string, args?: Record<string, any>): void {
|
||||
if (this.logLevel !== 'debug') return
|
||||
this.logger.debug(message, { ...args })
|
||||
}
|
||||
|
||||
end (): void {}
|
||||
|
||||
getParams (): ParamsType {
|
||||
|
||||
@@ -49,6 +49,12 @@ export interface Metrics extends MetricsData {
|
||||
opLog?: Record<string, OperationLog>
|
||||
}
|
||||
|
||||
/**
|
||||
* Root log verbosity for {@link MeasureContext.debug} (child contexts inherit unless overridden in {@link MeasureContext.newChild}).
|
||||
* @public
|
||||
*/
|
||||
export type MeasureLogLevel = 'info' | 'debug'
|
||||
|
||||
/**
|
||||
* @public
|
||||
*/
|
||||
@@ -58,6 +64,8 @@ export interface MeasureLogger {
|
||||
|
||||
warn: (message: string, obj?: Record<string, any>) => void
|
||||
|
||||
debug: (message: string, obj?: Record<string, any>) => void
|
||||
|
||||
logOperation: (operation: string, time: number, params: ParamsType) => void
|
||||
|
||||
childLogger?: (name: string, params: Record<string, any>) => MeasureLogger
|
||||
@@ -94,6 +102,7 @@ export interface MeasureContext<Q = any> {
|
||||
logger?: MeasureLogger
|
||||
span?: WithOptions['span'] // By default true
|
||||
meta?: Record<string, string | number | boolean>
|
||||
logLevel?: MeasureLogLevel
|
||||
}
|
||||
) => MeasureContext
|
||||
|
||||
@@ -128,6 +137,7 @@ export interface MeasureContext<Q = any> {
|
||||
error: (message: string, obj?: Record<string, any>) => void
|
||||
info: (message: string, obj?: Record<string, any>) => void
|
||||
warn: (message: string, obj?: Record<string, any>) => void
|
||||
debug: (message: string, obj?: Record<string, any>) => void // No-op unless this context was created with log level `debug`.
|
||||
|
||||
// Mark current context as complete
|
||||
// If no value is passed, time difference will be used.
|
||||
|
||||
@@ -150,6 +150,7 @@ const mockContext: MeasureContext = {
|
||||
info: jest.fn(),
|
||||
warn: jest.fn(),
|
||||
error: jest.fn(),
|
||||
debug: jest.fn(),
|
||||
with: jest.fn().mockImplementation((name, params, fn) => fn()),
|
||||
withSync: jest.fn().mockImplementation((name, params, fn) => fn()),
|
||||
measure: jest.fn(),
|
||||
|
||||
@@ -259,9 +259,14 @@ class PlatformQueueConsumerImpl implements ConsumerHandle {
|
||||
maxRetryDelay?: number // Maximum retry delay in seconds (default 10)
|
||||
}
|
||||
) {
|
||||
// Long handlers must call ConsumerControl.heartbeat(); these timeouts still help under broker/load jitter (e.g. Redpanda in Docker).
|
||||
const sessionTimeout = parseInt(process.env.KAFKA_CONSUMER_SESSION_TIMEOUT_MS ?? '90000', 10)
|
||||
this.cc = this.kafka.consumer({
|
||||
groupId: `${getKafkaTopicId(this.topic, this.config)}-${groupId}`,
|
||||
allowAutoTopicCreation: true
|
||||
allowAutoTopicCreation: true,
|
||||
sessionTimeout,
|
||||
rebalanceTimeout: Math.min(sessionTimeout * 2, 300000),
|
||||
heartbeatInterval: 3000
|
||||
})
|
||||
|
||||
void this.start().catch((err) => {
|
||||
|
||||
@@ -279,17 +279,20 @@ export function createModel (builder: Builder): void {
|
||||
{
|
||||
hidden: false,
|
||||
generated: false,
|
||||
allowedForAuthor: true,
|
||||
label: calendar.string.Reminder,
|
||||
group: calendar.ids.CalendarNotificationGroup,
|
||||
txClasses: [],
|
||||
// Scheduled reminders are created by the events-processor worker, but provider/type settings still expect a
|
||||
// tx class list. The notification doc itself is materialized via a direct createDoc, not by a tx trigger.
|
||||
txClasses: [core.class.TxCreateDoc],
|
||||
objectClass: calendar.class.Event,
|
||||
allowedForAuthor: true,
|
||||
onlyOwn: true,
|
||||
defaultEnabled: true,
|
||||
templates: {
|
||||
textTemplate: 'Reminder: {doc}',
|
||||
htmlTemplate: 'Reminder: {doc}',
|
||||
subjectTemplate: 'Reminder: {doc}'
|
||||
},
|
||||
defaultEnabled: false
|
||||
textTemplate: '{body}',
|
||||
htmlTemplate: '<p>{body}</p><p>{link}</p>',
|
||||
subjectTemplate: '{title}'
|
||||
}
|
||||
},
|
||||
calendar.ids.ReminderNotification
|
||||
)
|
||||
@@ -300,6 +303,12 @@ export function createModel (builder: Builder): void {
|
||||
enabledTypes: [calendar.ids.ReminderNotification]
|
||||
})
|
||||
|
||||
builder.createDoc(notification.class.NotificationProviderDefaults, core.space.Model, {
|
||||
provider: notification.providers.PushNotificationProvider,
|
||||
ignoredTypes: [],
|
||||
enabledTypes: [calendar.ids.ReminderNotification]
|
||||
})
|
||||
|
||||
builder.createDoc(
|
||||
activity.class.DocUpdateMessageViewlet,
|
||||
core.space.Model,
|
||||
|
||||
@@ -55,7 +55,6 @@ export default mergeIds(calendarId, calendar, {
|
||||
string: {
|
||||
ApplicationLabelCalendar: '' as IntlString,
|
||||
Event: '' as IntlString,
|
||||
Reminder: '' as IntlString,
|
||||
Shift: '' as IntlString,
|
||||
State: '' as IntlString,
|
||||
CreatedReminder: '' as IntlString,
|
||||
|
||||
@@ -257,6 +257,7 @@ const calendarPlugin = plugin(calendarId, {
|
||||
PersonsLabel: '' as IntlString,
|
||||
EventNumber: '' as IntlString,
|
||||
Reminders: '' as IntlString,
|
||||
Reminder: '' as IntlString,
|
||||
Today: '' as IntlString,
|
||||
Visibility: '' as IntlString,
|
||||
Public: '' as IntlString,
|
||||
|
||||
@@ -15,7 +15,7 @@
|
||||
<script lang="ts">
|
||||
import { Analytics } from '@hcengineering/analytics'
|
||||
import { AccessLevel, Calendar, generateEventId } from '@hcengineering/calendar'
|
||||
import { VisibilityEditor } from '@hcengineering/calendar-resources'
|
||||
import { EventReminders, VisibilityEditor } from '@hcengineering/calendar-resources'
|
||||
import calendar from '@hcengineering/calendar-resources/src/plugin'
|
||||
import { getCurrentEmployee } from '@hcengineering/contact'
|
||||
import core, { AttachedData, Doc, Ref, SortingOrder, generateId, getCurrentAccount } from '@hcengineering/core'
|
||||
@@ -104,7 +104,7 @@
|
||||
allDay: false,
|
||||
access: AccessLevel.Owner,
|
||||
visibility: todo.visibility === 'public' ? 'public' : 'freeBusy',
|
||||
reminders: [],
|
||||
reminders,
|
||||
user: myAccount.primarySocialId
|
||||
})
|
||||
Analytics.handleEvent(TimeEvents.ToDoScheduled, { id })
|
||||
@@ -126,6 +126,7 @@
|
||||
})
|
||||
|
||||
let slots: WorkSlot[] = []
|
||||
let reminders: number[] = []
|
||||
|
||||
function removeSlot (e: CustomEvent<{ _id: Ref<WorkSlot> }>): void {
|
||||
const index = slots.findIndex((p) => p._id === e.detail._id)
|
||||
@@ -151,7 +152,7 @@
|
||||
blockTime: true,
|
||||
access: AccessLevel.Owner,
|
||||
visibility: todo.visibility,
|
||||
reminders: [],
|
||||
reminders,
|
||||
calendar: _calendar,
|
||||
space: calendar.space.Calendar,
|
||||
_id: generateId(),
|
||||
@@ -267,6 +268,9 @@
|
||||
on:change={changeSlot}
|
||||
on:dueChange={changeDueSlot}
|
||||
/>
|
||||
{#if slots.length > 0}
|
||||
<EventReminders bind:reminders />
|
||||
{/if}
|
||||
</div>
|
||||
<div class="flex-row-reverse btn flex-no-shrink">
|
||||
<Button
|
||||
|
||||
@@ -13,7 +13,8 @@
|
||||
// limitations under the License.
|
||||
-->
|
||||
<script lang="ts">
|
||||
import calendar, { AccessLevel, Calendar, generateEventId } from '@hcengineering/calendar'
|
||||
import calendar, { AccessLevel, generateEventId } from '@hcengineering/calendar'
|
||||
import { EventReminders } from '@hcengineering/calendar-resources'
|
||||
import contact, { getCurrentEmployee } from '@hcengineering/contact'
|
||||
import { Ref, getCurrentAccount } from '@hcengineering/core'
|
||||
import { createQuery, getClient } from '@hcengineering/presentation'
|
||||
@@ -31,11 +32,39 @@
|
||||
const query = createQuery()
|
||||
|
||||
let slots: WorkSlot[] = []
|
||||
let reminders: number[] = []
|
||||
let remindersHydrated = false
|
||||
let remindersKey = ''
|
||||
let currentTodoId: Ref<ToDo> | undefined
|
||||
|
||||
$: if (todo?._id !== undefined && currentTodoId !== todo._id) {
|
||||
currentTodoId = todo._id
|
||||
remindersHydrated = false
|
||||
reminders = []
|
||||
remindersKey = ''
|
||||
}
|
||||
|
||||
$: query.query(time.class.WorkSlot, { attachedTo: todo._id }, (res) => {
|
||||
slots = res
|
||||
if (!remindersHydrated) {
|
||||
reminders = [...(slots[0]?.reminders ?? [])]
|
||||
remindersHydrated = true
|
||||
remindersKey = JSON.stringify(reminders)
|
||||
}
|
||||
})
|
||||
|
||||
$: {
|
||||
if (!remindersHydrated || slots.length === 0) {
|
||||
// no-op
|
||||
} else {
|
||||
const nextKey = JSON.stringify(reminders)
|
||||
if (nextKey !== remindersKey) {
|
||||
remindersKey = nextKey
|
||||
void syncRemindersForSlots()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async function change (e: CustomEvent<{ startDate: number, dueDate: number, slot: Ref<WorkSlot> }>): Promise<void> {
|
||||
const { startDate, dueDate, slot } = e.detail
|
||||
const workslot = slots.find((s) => s._id === slot)
|
||||
@@ -72,14 +101,22 @@
|
||||
access: AccessLevel.Owner,
|
||||
user: currentAccount.primarySocialId,
|
||||
visibility: todo.visibility === 'public' ? 'public' : 'freeBusy',
|
||||
reminders: []
|
||||
reminders
|
||||
})
|
||||
Analytics.handleEvent(TimeEvents.ToDoScheduled, { id: todo._id })
|
||||
}
|
||||
|
||||
async function syncRemindersForSlots (): Promise<void> {
|
||||
await Promise.all(
|
||||
slots.map(async (slot) => {
|
||||
await client.update(slot, { reminders })
|
||||
})
|
||||
)
|
||||
}
|
||||
|
||||
async function remove (e: CustomEvent<{ _id: Ref<WorkSlot> }>): Promise<void> {
|
||||
const object = slots.find((p) => p._id === e.detail._id)
|
||||
if (object) {
|
||||
if (object !== undefined) {
|
||||
showPopup(
|
||||
contact.component.DeleteConfirmationPopup,
|
||||
{
|
||||
@@ -98,4 +135,11 @@
|
||||
}
|
||||
</script>
|
||||
|
||||
<Workslots {slots} fixed={'toDo'} on:change={change} on:dueChange={dueChange} on:create={create} on:remove={remove} />
|
||||
<div class="flex-col">
|
||||
<Workslots {slots} fixed={'toDo'} on:change={change} on:dueChange={dueChange} on:create={create} on:remove={remove} />
|
||||
{#if slots.length > 0}
|
||||
<div class="flex pt-4">
|
||||
<EventReminders bind:reminders />
|
||||
</div>
|
||||
{/if}
|
||||
</div>
|
||||
|
||||
+21
-10
@@ -37,7 +37,7 @@ import { promisify } from 'util'
|
||||
import { gzip } from 'zlib'
|
||||
import { retrieveJson } from './utils'
|
||||
|
||||
import { unknownError } from '@hcengineering/platform'
|
||||
import platform, { PlatformError, unknownError } from '@hcengineering/platform'
|
||||
|
||||
export const COMMUNICATION_DOMAIN = 'communication' as OperationDomain
|
||||
interface RPCClientInfo {
|
||||
@@ -270,15 +270,26 @@ export function registerRPC (app: Express, sessions: SessionManager, ctx: Measur
|
||||
void withSession(req, res, 'tx', async (ctx, session, rateLimit) => {
|
||||
const tx: any = (await retrieveJson(req)) ?? {}
|
||||
|
||||
if (tx._class === core.class.TxDomainEvent) {
|
||||
const domainTx = tx as TxDomainEvent
|
||||
const { result } = await session.domainRequestRaw(ctx, domainTx.domain, {
|
||||
event: domainTx.event
|
||||
})
|
||||
await sendJson(req, res, result.value, rateLimitToHeaders(rateLimit))
|
||||
} else {
|
||||
const result = await session.txRaw(ctx, tx)
|
||||
await sendJson(req, res, result.result, rateLimitToHeaders(rateLimit))
|
||||
try {
|
||||
if (tx._class === core.class.TxDomainEvent) {
|
||||
const domainTx = tx as TxDomainEvent
|
||||
const { result } = await session.domainRequestRaw(ctx, domainTx.domain, {
|
||||
event: domainTx.event
|
||||
})
|
||||
await sendJson(req, res, result.value, rateLimitToHeaders(rateLimit))
|
||||
} else {
|
||||
const result = await session.txRaw(ctx, tx)
|
||||
await sendJson(req, res, result.result, rateLimitToHeaders(rateLimit))
|
||||
}
|
||||
} catch (err: unknown) {
|
||||
if (err instanceof PlatformError && err.status.code === platform.status.BadRequest) {
|
||||
sendError(res, 400, {
|
||||
message: 'Invalid tx',
|
||||
error: err.status
|
||||
})
|
||||
return
|
||||
}
|
||||
throw err
|
||||
}
|
||||
})
|
||||
})
|
||||
|
||||
@@ -2221,6 +2221,11 @@
|
||||
"projectFolder": "services/notification/pod-notification",
|
||||
"shouldPublish": false
|
||||
},
|
||||
{
|
||||
"packageName": "@hcengineering/pod-events-processor",
|
||||
"projectFolder": "services/notification/pod-events-processor",
|
||||
"shouldPublish": false
|
||||
},
|
||||
{
|
||||
"packageName": "@hcengineering/pod-telegram",
|
||||
"projectFolder": "services/telegram/pod-telegram",
|
||||
|
||||
@@ -44,6 +44,128 @@ import { QueueTopic, TriggerControl } from '@hcengineering/server-core'
|
||||
import { getHTMLPresenter, getTextPresenter } from '@hcengineering/server-notification-resources'
|
||||
import { generateToken } from '@hcengineering/server-token'
|
||||
|
||||
const scheduledNotificationTopic = 'scheduledNotification'
|
||||
|
||||
interface ScheduledNotificationMessage {
|
||||
kind: 'eventReminder'
|
||||
id: string
|
||||
eventId: Ref<Event>
|
||||
eventClass: Ref<Class<Event>>
|
||||
shiftMs: number
|
||||
targetDate: number
|
||||
}
|
||||
|
||||
type TimeMachineMessage =
|
||||
| {
|
||||
type: 'schedule'
|
||||
id: string
|
||||
targetDate: number
|
||||
topic: string
|
||||
data: ScheduledNotificationMessage
|
||||
}
|
||||
| {
|
||||
type: 'cancel'
|
||||
id: string
|
||||
}
|
||||
|
||||
type TimeMachineScheduleMessage = Extract<TimeMachineMessage, { type: 'schedule' }>
|
||||
|
||||
function eventReminderPrefix (eventId: Ref<Event>): string {
|
||||
return `eventReminder_${eventId}_`
|
||||
}
|
||||
|
||||
function eventReminderTimerId (eventId: Ref<Event>, shiftMs: number): string {
|
||||
// Stable so we can cancel by `${prefix}%`.
|
||||
return `${eventReminderPrefix(eventId)}${shiftMs}`
|
||||
}
|
||||
|
||||
async function cancelEventReminders (control: TriggerControl, eventId: Ref<Event>): Promise<void> {
|
||||
try {
|
||||
const queue = control.queue
|
||||
if (queue === undefined) return
|
||||
const producer = queue.getProducer<TimeMachineMessage>(control.ctx, QueueTopic.TimeMachine)
|
||||
const cancelId = `${eventReminderPrefix(eventId)}%`
|
||||
await producer.send(control.ctx, control.workspace.uuid, [{ type: 'cancel', id: cancelId }])
|
||||
control.ctx.info('Queued event reminder cancel', {
|
||||
queueTopic: QueueTopic.TimeMachine,
|
||||
eventId,
|
||||
timerIdPattern: cancelId
|
||||
})
|
||||
} catch (err) {
|
||||
control.ctx.error('Failed to cancel Event reminders', { err, eventId })
|
||||
}
|
||||
}
|
||||
|
||||
async function scheduleEventReminders (control: TriggerControl, eventId: Ref<Event>): Promise<void> {
|
||||
try {
|
||||
const queue = control.queue
|
||||
if (queue === undefined) return
|
||||
|
||||
const event = (await control.findAll(control.ctx, calendar.class.Event, { _id: eventId }, { limit: 1 }))[0]
|
||||
if (event === undefined) return
|
||||
|
||||
// Reset existing timers for this Event on any relevant change.
|
||||
await cancelEventReminders(control, eventId)
|
||||
|
||||
const reminders = event.reminders ?? []
|
||||
if (reminders.length === 0) return
|
||||
|
||||
const now = Date.now()
|
||||
const msgs: TimeMachineScheduleMessage[] = []
|
||||
|
||||
for (const shiftMs of reminders) {
|
||||
if (typeof shiftMs !== 'number' || Number.isNaN(shiftMs)) continue
|
||||
// `shiftMs` is the positive offset before the event (in ms), matching the convention used by
|
||||
// ReminderPopup and pod-calendar Google export.
|
||||
const targetDate = event.date - shiftMs
|
||||
if (targetDate <= now) continue
|
||||
|
||||
const id = eventReminderTimerId(eventId, shiftMs)
|
||||
const data: ScheduledNotificationMessage = {
|
||||
kind: 'eventReminder',
|
||||
id,
|
||||
eventId,
|
||||
eventClass: event._class,
|
||||
shiftMs,
|
||||
targetDate
|
||||
}
|
||||
msgs.push({
|
||||
type: 'schedule',
|
||||
id,
|
||||
targetDate,
|
||||
topic: scheduledNotificationTopic,
|
||||
data
|
||||
})
|
||||
}
|
||||
|
||||
if (msgs.length === 0) {
|
||||
control.ctx.info('Skipped event reminder scheduling', {
|
||||
queueTopic: QueueTopic.TimeMachine,
|
||||
eventId,
|
||||
eventClass: event._class,
|
||||
reminderCount: reminders.length,
|
||||
reason: 'no-future-reminders'
|
||||
})
|
||||
return
|
||||
}
|
||||
const producer = queue.getProducer<TimeMachineMessage>(control.ctx, QueueTopic.TimeMachine)
|
||||
await producer.send(control.ctx, control.workspace.uuid, msgs)
|
||||
control.ctx.info('Queued event reminders', {
|
||||
queueTopic: QueueTopic.TimeMachine,
|
||||
eventId,
|
||||
eventClass: event._class,
|
||||
reminderCount: reminders.length,
|
||||
enqueuedCount: msgs.length,
|
||||
timerIds: msgs.map((msg) => msg.id),
|
||||
targetDates: msgs.map((msg) => msg.targetDate)
|
||||
})
|
||||
} catch (err) {
|
||||
control.ctx.error('Failed to schedule Event reminders', { err, eventId })
|
||||
}
|
||||
}
|
||||
|
||||
export { scheduleEventReminders, cancelEventReminders }
|
||||
|
||||
/**
|
||||
* @public
|
||||
*/
|
||||
@@ -214,6 +336,10 @@ async function onEventUpdate (ctx: TxUpdateDoc<Event>, control: TriggerControl):
|
||||
void sendEventToService(event, 'update', control)
|
||||
}
|
||||
void putEventToQueue(control, 'update', event, ctx.modifiedBy, ops)
|
||||
// Reschedule reminders if the event start time or the reminder offsets changed.
|
||||
if (ops.date !== undefined || ops.reminders !== undefined) {
|
||||
void scheduleEventReminders(control, ctx.objectId)
|
||||
}
|
||||
if (event.access !== 'owner') return []
|
||||
const events = await control.findAll(control.ctx, calendar.class.Event, { eventId: event.eventId })
|
||||
const res: Tx[] = []
|
||||
@@ -363,6 +489,8 @@ async function onEventCreate (ctx: TxCreateDoc<Event>, control: TriggerControl):
|
||||
void sendEventToService(event, 'create', control)
|
||||
}
|
||||
void putEventToQueue(control, 'create', event, ctx.modifiedBy)
|
||||
// Schedule reminders for any newly created event (including WorkSlots, since those are Events too).
|
||||
void scheduleEventReminders(control, event._id)
|
||||
if (event.access !== 'owner') return []
|
||||
const res: Tx[] = []
|
||||
const { _class, space, ...attr } = event
|
||||
@@ -398,6 +526,7 @@ async function onEventCreate (ctx: TxCreateDoc<Event>, control: TriggerControl):
|
||||
async function onRemoveEvent (ctx: TxRemoveDoc<Event>, control: TriggerControl): Promise<Tx[]> {
|
||||
const removed = control.removedMap.get(ctx.objectId) as Event
|
||||
const res: Tx[] = []
|
||||
void cancelEventReminders(control, ctx.objectId)
|
||||
if (removed !== undefined) {
|
||||
if (ctx.modifiedBy !== core.account.System && removed.access === 'owner') {
|
||||
void sendEventToService(removed, 'delete', control)
|
||||
|
||||
@@ -0,0 +1,243 @@
|
||||
//
|
||||
// Copyright © 2026 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.
|
||||
//
|
||||
|
||||
/* eslint-disable @typescript-eslint/consistent-type-assertions */
|
||||
|
||||
import calendar, { type Event } from '@hcengineering/calendar'
|
||||
import core, { generateId } from '@hcengineering/core'
|
||||
import type { PlatformQueueProducer } from '@hcengineering/server-core'
|
||||
import { cancelEventReminders, scheduleEventReminders } from './index'
|
||||
|
||||
type AnyProducer = PlatformQueueProducer<any>
|
||||
|
||||
function makeQueueMock (): { queue: { getProducer: jest.Mock }, send: jest.Mock } {
|
||||
const send = jest.fn(async () => {})
|
||||
const producer: AnyProducer = { send, close: async () => {}, getQueue: () => queue as any } as any
|
||||
const queue = {
|
||||
getProducer: jest.fn(() => producer)
|
||||
}
|
||||
return { queue, send }
|
||||
}
|
||||
|
||||
function makeControl (overrides: Partial<any> = {}): { control: any, send: jest.Mock } {
|
||||
const { queue, send } = makeQueueMock()
|
||||
|
||||
const control: any = {
|
||||
ctx: {
|
||||
error: jest.fn(),
|
||||
warn: jest.fn(),
|
||||
info: jest.fn(),
|
||||
contextData: { account: { uuid: generateId(), primarySocialId: core.account.System } }
|
||||
},
|
||||
workspace: { uuid: generateId(), url: 'ws', dataId: 'ws' },
|
||||
hierarchy: {
|
||||
isDerived: jest.fn(() => true),
|
||||
classHierarchyMixin: jest.fn(() => undefined)
|
||||
},
|
||||
modelDb: { findAll: jest.fn(), findAllSync: jest.fn(), getObject: jest.fn() },
|
||||
removedMap: new Map(),
|
||||
userStatusMap: new Map(),
|
||||
queue,
|
||||
cache: new Map(),
|
||||
contextCache: new Map(),
|
||||
storageAdapter: {} as any,
|
||||
serviceAdaptersManager: {} as any,
|
||||
lowLevel: {} as any,
|
||||
txFactory: { createTxUpdateDoc: jest.fn(), createTxRemoveDoc: jest.fn(), createTxCollectionCUD: jest.fn() } as any,
|
||||
apply: jest.fn(async () => ({})),
|
||||
domainRequest: jest.fn(async () => ({})),
|
||||
queryFind: jest.fn(async () => []),
|
||||
txes: [],
|
||||
findAll: jest.fn(async () => [])
|
||||
}
|
||||
|
||||
Object.assign(control, overrides)
|
||||
|
||||
return { control, send }
|
||||
}
|
||||
|
||||
describe('event reminder scheduling (TimeMachine)', () => {
|
||||
const eventClass = calendar.class.Event
|
||||
|
||||
it('schedules reminders at `event.date - shiftMs` and uses the eventReminder_ timer prefix', async () => {
|
||||
const eventId = generateId() as any
|
||||
// 1 hour in the future, so a 5-minute "before" reminder is in the future too.
|
||||
const eventDate = Date.now() + 60 * 60_000
|
||||
const shiftMs = 5 * 60_000
|
||||
|
||||
const { control, send } = makeControl()
|
||||
;(control.findAll as jest.Mock).mockImplementation(async (_ctx: any, _class: any, query: any) => {
|
||||
if (_class === calendar.class.Event && query?._id === eventId) {
|
||||
return [
|
||||
{
|
||||
_id: eventId,
|
||||
_class: eventClass,
|
||||
space: core.space.Workspace,
|
||||
date: eventDate,
|
||||
dueDate: eventDate + 60_000,
|
||||
reminders: [shiftMs]
|
||||
} as Partial<Event>
|
||||
]
|
||||
}
|
||||
return []
|
||||
})
|
||||
|
||||
await scheduleEventReminders(control, eventId)
|
||||
|
||||
// Producer.send signature: (ctx, workspace, msgs).
|
||||
const msgs = send.mock.calls.map((c: any[]) => c[2]).flat()
|
||||
const cancelMsg = msgs.find((m: any) => m.type === 'cancel')
|
||||
const scheduleMsg = msgs.find((m: any) => m.type === 'schedule' && m.topic === 'scheduledNotification')
|
||||
|
||||
expect(cancelMsg).toBeDefined()
|
||||
expect(cancelMsg.id).toBe(`eventReminder_${eventId}_%`)
|
||||
|
||||
expect(scheduleMsg).toBeDefined()
|
||||
expect(scheduleMsg.id).toBe(`eventReminder_${eventId}_${shiftMs}`)
|
||||
// Reminder must fire BEFORE the event, exactly `shiftMs` earlier.
|
||||
expect(scheduleMsg.targetDate).toBe(eventDate - shiftMs)
|
||||
expect(scheduleMsg.data.kind).toBe('eventReminder')
|
||||
expect(scheduleMsg.data.eventId).toBe(eventId)
|
||||
expect(scheduleMsg.data.eventClass).toBe(eventClass)
|
||||
expect(scheduleMsg.data.shiftMs).toBe(shiftMs)
|
||||
expect(scheduleMsg.data.targetDate).toBe(eventDate - shiftMs)
|
||||
})
|
||||
|
||||
it('skips reminders that resolve to the past', async () => {
|
||||
const eventId = generateId() as any
|
||||
// Event 1 minute in the future, 5-min reminder lands 4 minutes in the past — must be skipped.
|
||||
const eventDate = Date.now() + 60_000
|
||||
const shiftMs = 5 * 60_000
|
||||
|
||||
const { control, send } = makeControl()
|
||||
;(control.findAll as jest.Mock).mockImplementation(async (_ctx: any, _class: any, query: any) => {
|
||||
if (_class === calendar.class.Event && query?._id === eventId) {
|
||||
return [
|
||||
{
|
||||
_id: eventId,
|
||||
_class: eventClass,
|
||||
space: core.space.Workspace,
|
||||
date: eventDate,
|
||||
dueDate: eventDate + 60_000,
|
||||
reminders: [shiftMs]
|
||||
} as Partial<Event>
|
||||
]
|
||||
}
|
||||
return []
|
||||
})
|
||||
|
||||
await scheduleEventReminders(control, eventId)
|
||||
|
||||
const msgs = send.mock.calls.map((c: any[]) => c[2]).flat()
|
||||
expect(msgs.find((m: any) => m.type === 'schedule')).toBeUndefined()
|
||||
// Cancel for the prefix is still issued so any prior timers get cleared.
|
||||
expect(msgs.find((m: any) => m.type === 'cancel')).toBeDefined()
|
||||
})
|
||||
|
||||
it('does not schedule when the event has no reminders configured', async () => {
|
||||
const eventId = generateId() as any
|
||||
|
||||
const { control, send } = makeControl()
|
||||
;(control.findAll as jest.Mock).mockImplementation(async (_ctx: any, _class: any, query: any) => {
|
||||
if (_class === calendar.class.Event && query?._id === eventId) {
|
||||
return [
|
||||
{
|
||||
_id: eventId,
|
||||
_class: eventClass,
|
||||
space: core.space.Workspace,
|
||||
date: Date.now() + 60 * 60_000,
|
||||
dueDate: Date.now() + 70 * 60_000,
|
||||
reminders: []
|
||||
} as Partial<Event>
|
||||
]
|
||||
}
|
||||
return []
|
||||
})
|
||||
|
||||
await scheduleEventReminders(control, eventId)
|
||||
|
||||
const msgs = send.mock.calls.map((c: any[]) => c[2]).flat()
|
||||
// Cancel always issued (resets prior state); but no schedule msg.
|
||||
expect(msgs.find((m: any) => m.type === 'schedule')).toBeUndefined()
|
||||
})
|
||||
|
||||
it('schedules ALL future reminders when more than one is configured', async () => {
|
||||
const eventId = generateId() as any
|
||||
const eventDate = Date.now() + 60 * 60_000
|
||||
const shifts = [5 * 60_000, 15 * 60_000, 30 * 60_000]
|
||||
|
||||
const { control, send } = makeControl()
|
||||
;(control.findAll as jest.Mock).mockImplementation(async (_ctx: any, _class: any, query: any) => {
|
||||
if (_class === calendar.class.Event && query?._id === eventId) {
|
||||
return [
|
||||
{
|
||||
_id: eventId,
|
||||
_class: eventClass,
|
||||
space: core.space.Workspace,
|
||||
date: eventDate,
|
||||
dueDate: eventDate + 60_000,
|
||||
reminders: shifts
|
||||
} as Partial<Event>
|
||||
]
|
||||
}
|
||||
return []
|
||||
})
|
||||
|
||||
await scheduleEventReminders(control, eventId)
|
||||
|
||||
const msgs = send.mock.calls.map((c: any[]) => c[2]).flat()
|
||||
const scheduleMsgs = msgs.filter((m: any) => m.type === 'schedule')
|
||||
expect(scheduleMsgs).toHaveLength(shifts.length)
|
||||
for (const shiftMs of shifts) {
|
||||
const m = scheduleMsgs.find((s: any) => s.id === `eventReminder_${eventId}_${shiftMs}`)
|
||||
expect(m).toBeDefined()
|
||||
expect(m.targetDate).toBe(eventDate - shiftMs)
|
||||
}
|
||||
})
|
||||
|
||||
it('cancelEventReminders sends a wildcard cancel for the prefix', async () => {
|
||||
const eventId = generateId() as any
|
||||
const { control, send } = makeControl()
|
||||
|
||||
await cancelEventReminders(control, eventId)
|
||||
|
||||
const msgs = send.mock.calls.map((c: any[]) => c[2]).flat()
|
||||
const cancelMsg = msgs.find((m: any) => m.type === 'cancel')
|
||||
expect(cancelMsg).toBeDefined()
|
||||
expect(cancelMsg.id).toBe(`eventReminder_${eventId}_%`)
|
||||
})
|
||||
|
||||
it('does nothing when control.queue is undefined', async () => {
|
||||
const eventId = generateId() as any
|
||||
const { control, send } = makeControl({ queue: undefined })
|
||||
|
||||
await scheduleEventReminders(control, eventId)
|
||||
await cancelEventReminders(control, eventId)
|
||||
|
||||
expect(send).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('does nothing when the event is not found', async () => {
|
||||
const eventId = generateId() as any
|
||||
const { control, send } = makeControl()
|
||||
;(control.findAll as jest.Mock).mockResolvedValue([])
|
||||
|
||||
await scheduleEventReminders(control, eventId)
|
||||
|
||||
// No event to schedule for — only the cancel-on-reset behavior is skipped too because we bail
|
||||
// before that. Verify nothing was sent.
|
||||
expect(send).not.toHaveBeenCalled()
|
||||
})
|
||||
})
|
||||
@@ -15,6 +15,7 @@
|
||||
|
||||
import serverCore, { TriggerControl } from '@hcengineering/server-core'
|
||||
import serverNotification, { PUSH_NOTIFICATION_TITLE_SIZE } from '@hcengineering/server-notification'
|
||||
import type { ReceiverInfo } from '@hcengineering/server-notification'
|
||||
import {
|
||||
AccountUuid,
|
||||
Class,
|
||||
@@ -49,6 +50,7 @@ import contact, {
|
||||
} from '@hcengineering/contact'
|
||||
import { AvailableProvidersCache, AvailableProvidersCacheKey, getTranslatedNotificationContent } from './index'
|
||||
import { getPerson } from '@hcengineering/server-contact'
|
||||
import { getAllowedProviders, getNotificationProviderControl, getReceiversInfo } from './utils'
|
||||
|
||||
async function createPushFromInbox (
|
||||
control: TriggerControl,
|
||||
@@ -235,25 +237,51 @@ export async function PushNotificationsHandler (
|
||||
): Promise<Tx[]> {
|
||||
const availableProviders: AvailableProvidersCache = control.contextCache.get(AvailableProvidersCacheKey) ?? new Map()
|
||||
|
||||
const all: InboxNotification[] = txes
|
||||
.map((tx) => TxProcessor.createDoc2Doc(tx))
|
||||
.filter(
|
||||
(it) =>
|
||||
availableProviders.get(it._id)?.find((p) => p === notification.providers.PushNotificationProvider) !== undefined
|
||||
)
|
||||
const all: InboxNotification[] = txes.map((tx) => TxProcessor.createDoc2Doc(tx))
|
||||
|
||||
if (all.length === 0) {
|
||||
// First pass: use cache if present.
|
||||
const pushEnabled: InboxNotification[] = all.filter(
|
||||
(it) =>
|
||||
availableProviders.get(it._id)?.find((p) => p === notification.providers.PushNotificationProvider) !== undefined
|
||||
)
|
||||
|
||||
// Fallback: if cache doesn't have the provider info (e.g. scheduled notifications created outside tx-trigger paths),
|
||||
// compute allowed providers from notification type + user settings.
|
||||
if (pushEnabled.length < all.length) {
|
||||
const notificationControl = await getNotificationProviderControl(control.ctx, control)
|
||||
const receivers: ReceiverInfo[] = await getReceiversInfo(control.ctx, [...new Set(all.map((n) => n.user))], control)
|
||||
const receiverByAccount = new Map(receivers.map((r) => [r.account, r]))
|
||||
|
||||
for (const n of all) {
|
||||
if (availableProviders.get(n._id) !== undefined) continue
|
||||
if (pushEnabled.includes(n)) continue
|
||||
|
||||
const type = (n.types ?? [])[0]
|
||||
if (type === undefined) continue
|
||||
|
||||
const notificationType = control.modelDb.getObject(type)
|
||||
const receiver = receiverByAccount.get(n.user)
|
||||
if (receiver === undefined) continue
|
||||
|
||||
const allowedProviders = getAllowedProviders(control, receiver.socialIds, notificationType, notificationControl)
|
||||
if (allowedProviders.includes(notification.providers.PushNotificationProvider)) {
|
||||
pushEnabled.push(n)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (pushEnabled.length === 0) {
|
||||
return []
|
||||
}
|
||||
|
||||
const receivers = new Set(all.map((it) => it.user))
|
||||
const receivers = new Set(pushEnabled.map((it) => it.user))
|
||||
const subscriptions = (await control.queryFind(control.ctx, notification.class.PushSubscription, {})).filter((it) =>
|
||||
receivers.has(it.user)
|
||||
)
|
||||
|
||||
const res: Tx[] = []
|
||||
|
||||
for (const inboxNotification of all) {
|
||||
for (const inboxNotification of pushEnabled) {
|
||||
const { user } = inboxNotification
|
||||
const userSubscriptions = subscriptions.filter((it) => it.user === user)
|
||||
|
||||
|
||||
@@ -32,6 +32,7 @@ describe('AttachmentHandler', () => {
|
||||
info: jest.fn(),
|
||||
error: jest.fn(),
|
||||
warn: jest.fn(),
|
||||
debug: jest.fn(),
|
||||
logOperation: jest.fn(),
|
||||
childLogger: jest.fn(),
|
||||
close: jest.fn()
|
||||
@@ -40,6 +41,7 @@ describe('AttachmentHandler', () => {
|
||||
error: jest.fn(),
|
||||
info: jest.fn(),
|
||||
warn: jest.fn(),
|
||||
debug: jest.fn(),
|
||||
end: jest.fn(),
|
||||
getParams: jest.fn()
|
||||
}
|
||||
|
||||
@@ -0,0 +1,8 @@
|
||||
module.exports = {
|
||||
extends: ['./node_modules/@hcengineering/platform-rig/profiles/default/eslint.config.json'],
|
||||
parserOptions: {
|
||||
tsconfigRootDir: __dirname,
|
||||
project: './tsconfig.json'
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,6 @@
|
||||
FROM hardcoreeng/base-slim:v20250916
|
||||
WORKDIR /usr/src/app
|
||||
|
||||
COPY bundle/bundle.js ./
|
||||
|
||||
CMD [ "node", "bundle.js" ]
|
||||
@@ -0,0 +1,8 @@
|
||||
module.exports = {
|
||||
preset: 'ts-jest',
|
||||
testEnvironment: 'node',
|
||||
testMatch: ['**/?(*.)+(spec|test).[jt]s?(x)'],
|
||||
roots: ['./src'],
|
||||
coverageReporters: ['text-summary', 'html']
|
||||
}
|
||||
|
||||
@@ -0,0 +1,71 @@
|
||||
{
|
||||
"name": "@hcengineering/pod-events-processor",
|
||||
"version": "0.7.0",
|
||||
"main": "lib/index.js",
|
||||
"svelte": "src/index.ts",
|
||||
"types": "types/index.d.ts",
|
||||
"files": [
|
||||
"lib/**/*",
|
||||
"types/**/*",
|
||||
"tsconfig.json"
|
||||
],
|
||||
"author": "Hardcore Engineering Inc.",
|
||||
"scripts": {
|
||||
"build": "compile",
|
||||
"build:watch": "compile",
|
||||
"test": "jest --passWithNoTests --silent",
|
||||
"_phase:bundle": "rushx bundle",
|
||||
"_phase:docker-build": "rushx docker:build",
|
||||
"_phase:docker-staging": "rushx docker:staging",
|
||||
"bundle": "node ../../../common/scripts/esbuild.js --external=ws",
|
||||
"docker:build": "../../../common/scripts/docker_build.sh hardcoreeng/events-processor",
|
||||
"docker:staging": "../../../common/scripts/docker_tag.sh hardcoreeng/events-processor staging",
|
||||
"docker:push": "../../../common/scripts/docker_tag.sh hardcoreeng/events-processor",
|
||||
"run-local": "cross-env ts-node src/index.ts",
|
||||
"format": "format src",
|
||||
"_phase:build": "compile transpile src",
|
||||
"_phase:test": "jest --passWithNoTests --silent",
|
||||
"_phase:format": "format src",
|
||||
"_phase:validate": "compile validate"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@hcengineering/platform-rig": "workspace:^0.7.21",
|
||||
"@tsconfig/node16": "^1.0.4",
|
||||
"@types/node": "^22.18.1",
|
||||
"@typescript-eslint/eslint-plugin": "^6.21.0",
|
||||
"@types/jest": "^29.5.5",
|
||||
"@typescript-eslint/parser": "^6.21.0",
|
||||
"cross-env": "~7.0.3",
|
||||
"esbuild": "^0.25.10",
|
||||
"eslint": "^8.54.0",
|
||||
"eslint-config-standard-with-typescript": "^40.0.0",
|
||||
"eslint-plugin-import": "^2.26.0",
|
||||
"eslint-plugin-n": "^15.4.0",
|
||||
"eslint-plugin-node": "^11.1.0",
|
||||
"eslint-plugin-promise": "^6.1.1",
|
||||
"jest": "^29.7.0",
|
||||
"prettier": "^3.6.2",
|
||||
"ts-jest": "^29.1.1",
|
||||
"ts-node": "^10.9.2",
|
||||
"typescript": "^5.9.3"
|
||||
},
|
||||
"dependencies": {
|
||||
"@hcengineering/account-client": "workspace:^0.7.25",
|
||||
"@hcengineering/analytics": "workspace:^0.7.19",
|
||||
"@hcengineering/analytics-service": "workspace:^0.7.19",
|
||||
"@hcengineering/api-client": "workspace:^0.7.25",
|
||||
"@hcengineering/calendar": "workspace:^0.7.0",
|
||||
"@hcengineering/contact": "workspace:^0.7.0",
|
||||
"@hcengineering/core": "workspace:^0.7.26",
|
||||
"@hcengineering/kafka": "workspace:^0.7.18",
|
||||
"@hcengineering/notification": "workspace:^0.7.0",
|
||||
"@hcengineering/platform": "workspace:^0.7.20",
|
||||
"@hcengineering/server-client": "workspace:^0.7.16",
|
||||
"@hcengineering/server-core": "workspace:^0.7.19",
|
||||
"@hcengineering/server-token": "workspace:^0.7.18",
|
||||
"@hcengineering/text-core": "workspace:^0.7.19",
|
||||
"@hcengineering/time": "workspace:^0.7.0",
|
||||
"dotenv": "^16.4.5"
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,77 @@
|
||||
//
|
||||
// Copyright © 2026 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 { PersonId, WorkspaceUuid } from '@hcengineering/core'
|
||||
import type { ClientBundle } from '../client'
|
||||
import {
|
||||
clearClientCachesForTests,
|
||||
clearInFlightClientCreation,
|
||||
getCacheKey,
|
||||
getCachedClient,
|
||||
getInFlightClientCreation,
|
||||
setCachedClient,
|
||||
setInFlightClientCreation
|
||||
} from '../clientCache'
|
||||
import config from '../config'
|
||||
|
||||
describe('clientCache', () => {
|
||||
const workspace = 'workspace-1' as WorkspaceUuid
|
||||
const serviceTag = 'events-processor'
|
||||
const socialId = 'person-1' as PersonId
|
||||
|
||||
const client = {} as unknown as ClientBundle['client']
|
||||
const accountClient = {} as unknown as ClientBundle['accountClient']
|
||||
const bundle: ClientBundle = { client, accountClient }
|
||||
|
||||
beforeEach(() => {
|
||||
clearClientCachesForTests()
|
||||
})
|
||||
|
||||
afterEach(() => {
|
||||
jest.restoreAllMocks()
|
||||
})
|
||||
|
||||
it('uses system marker in cache key when socialId is undefined', () => {
|
||||
expect(getCacheKey(workspace, undefined, serviceTag)).toBe('workspace-1:system:events-processor')
|
||||
})
|
||||
|
||||
it('returns cached value before ttl expires', () => {
|
||||
const now = 1_000
|
||||
jest.spyOn(Date, 'now').mockReturnValue(now)
|
||||
|
||||
setCachedClient(workspace, socialId, serviceTag, bundle)
|
||||
expect(getCachedClient(workspace, socialId, serviceTag)).toBe(bundle)
|
||||
})
|
||||
|
||||
it('evicts cached value after ttl expires', () => {
|
||||
const now = 1_000
|
||||
const dateNow = jest.spyOn(Date, 'now').mockReturnValue(now)
|
||||
setCachedClient(workspace, socialId, serviceTag, bundle)
|
||||
|
||||
dateNow.mockReturnValue(now + config.ClientCacheTtlMs + 1)
|
||||
expect(getCachedClient(workspace, socialId, serviceTag)).toBeUndefined()
|
||||
})
|
||||
|
||||
it('stores and clears in-flight client creation', () => {
|
||||
const key = getCacheKey(workspace, socialId, serviceTag)
|
||||
const creation = Promise.resolve(bundle)
|
||||
|
||||
setInFlightClientCreation(key, creation)
|
||||
expect(getInFlightClientCreation(key)).toBe(creation)
|
||||
|
||||
clearInFlightClientCreation(key)
|
||||
expect(getInFlightClientCreation(key)).toBeUndefined()
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,579 @@
|
||||
//
|
||||
// Copyright © 2026 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 calendar from '@hcengineering/calendar'
|
||||
import contact from '@hcengineering/contact'
|
||||
import core, { type MeasureContext, type WorkspaceUuid } from '@hcengineering/core'
|
||||
import notification from '@hcengineering/notification'
|
||||
import { PlatformError, Severity, Status } from '@hcengineering/platform'
|
||||
import type { ConsumerControl } from '@hcengineering/server-core'
|
||||
import time from '@hcengineering/time'
|
||||
import { getClient } from '../client'
|
||||
import type { ScheduledNotificationMessage } from '../types'
|
||||
import { buildReminderNotificationId, handleScheduledNotification } from '../worker'
|
||||
|
||||
const eventClassRef = 'calendar:class:Event' as any
|
||||
|
||||
jest.mock('../client', () => ({
|
||||
getClient: jest.fn()
|
||||
}))
|
||||
|
||||
describe('handleScheduledNotification', () => {
|
||||
const workspaceUuid = 'workspace-1' as WorkspaceUuid
|
||||
|
||||
const control = {
|
||||
heartbeat: jest.fn(async () => {})
|
||||
} as unknown as ConsumerControl
|
||||
|
||||
const ctx = {
|
||||
info: jest.fn(),
|
||||
error: jest.fn(),
|
||||
warn: jest.fn()
|
||||
} as unknown as MeasureContext
|
||||
|
||||
const findOne = jest.fn()
|
||||
const createDoc = jest.fn()
|
||||
const findPersonBySocialId = jest.fn()
|
||||
const isDerived = jest.fn((child: any, parent: any) => {
|
||||
if (child === parent) return true
|
||||
if (parent === time.class.ToDo) {
|
||||
// ToDo and its known subclasses.
|
||||
return child === time.class.ToDo || child === 'time:class:ProjectToDo'
|
||||
}
|
||||
return false
|
||||
})
|
||||
const getHierarchy = jest.fn(() => ({ isDerived }))
|
||||
const client = { findOne, createDoc, getHierarchy }
|
||||
const accountClient = { findPersonBySocialId }
|
||||
|
||||
beforeEach(() => {
|
||||
jest.clearAllMocks()
|
||||
;(getClient as jest.Mock).mockResolvedValue({ client, accountClient })
|
||||
})
|
||||
|
||||
// --------------------------------------------------------------------------------
|
||||
// Common helpers
|
||||
// --------------------------------------------------------------------------------
|
||||
const workSlotMessage: ScheduledNotificationMessage = {
|
||||
kind: 'eventReminder',
|
||||
id: 'timer-workslot-1',
|
||||
eventId: 'workslot-1' as any,
|
||||
eventClass: time.class.WorkSlot as any,
|
||||
shiftMs: 1000,
|
||||
targetDate: 1_000_000
|
||||
}
|
||||
const eventMessage: ScheduledNotificationMessage = {
|
||||
kind: 'eventReminder',
|
||||
id: 'timer-event-1',
|
||||
eventId: 'event-1' as any,
|
||||
eventClass: eventClassRef,
|
||||
shiftMs: 1000,
|
||||
targetDate: 1_000_000
|
||||
}
|
||||
|
||||
// --------------------------------------------------------------------------------
|
||||
// Negative paths: things the worker should silently ignore.
|
||||
// --------------------------------------------------------------------------------
|
||||
it('returns early for non-eventReminder messages (e.g. legacy todoReminder messages)', async () => {
|
||||
const legacy = { ...workSlotMessage, kind: 'todoReminder' as any }
|
||||
await handleScheduledNotification(ctx, workspaceUuid, legacy as any, control)
|
||||
|
||||
expect(getClient).not.toHaveBeenCalled()
|
||||
expect(control.heartbeat).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
// --------------------------------------------------------------------------------
|
||||
// WorkSlot path: notification points at the parent ToDo (legacy UX preserved).
|
||||
// --------------------------------------------------------------------------------
|
||||
describe('WorkSlot/ToDo branch', () => {
|
||||
const expectedNotificationId = buildReminderNotificationId(workSlotMessage.id)
|
||||
|
||||
it('skips when the ToDo is already done', async () => {
|
||||
findOne.mockImplementation(async (klass: any) => {
|
||||
if (klass === time.class.WorkSlot) {
|
||||
return {
|
||||
_id: workSlotMessage.eventId,
|
||||
_class: time.class.WorkSlot,
|
||||
space: 'space-1',
|
||||
attachedTo: 'todo-1',
|
||||
attachedToClass: time.class.ToDo
|
||||
}
|
||||
}
|
||||
if (klass === time.class.ToDo) {
|
||||
return {
|
||||
_id: 'todo-1',
|
||||
_class: 'time:class:ToDo',
|
||||
space: 'space-1',
|
||||
user: 'employee-1',
|
||||
title: 'Done todo',
|
||||
doneOn: Date.now()
|
||||
}
|
||||
}
|
||||
return undefined
|
||||
})
|
||||
|
||||
await handleScheduledNotification(ctx, workspaceUuid, workSlotMessage, control)
|
||||
|
||||
expect(createDoc).not.toHaveBeenCalled()
|
||||
expect(ctx.info).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('creates DocNotifyContext + CommonInboxNotification pointing at the ToDo', async () => {
|
||||
let docNotifyContextCreated = false
|
||||
findOne.mockImplementation(async (klass: any, query: any) => {
|
||||
if (klass === time.class.WorkSlot) {
|
||||
return {
|
||||
_id: workSlotMessage.eventId,
|
||||
_class: time.class.WorkSlot,
|
||||
space: 'space-1',
|
||||
attachedTo: 'todo-1',
|
||||
attachedToClass: time.class.ToDo
|
||||
}
|
||||
}
|
||||
if (klass === time.class.ToDo) {
|
||||
return {
|
||||
_id: 'todo-1',
|
||||
_class: 'time:class:ToDo',
|
||||
space: 'space-1',
|
||||
user: 'employee-1',
|
||||
title: 'Todo title',
|
||||
doneOn: null
|
||||
}
|
||||
}
|
||||
if (klass === contact.mixin.Employee) return { personUuid: 'person-1' }
|
||||
if (klass === contact.class.PersonSpace) return { _id: 'person-space-1' }
|
||||
if (klass === notification.class.CommonInboxNotification && query?._id === expectedNotificationId) {
|
||||
return undefined
|
||||
}
|
||||
if (klass === notification.class.DocNotifyContext) {
|
||||
return docNotifyContextCreated ? { _id: 'doc-notify-created-id' } : undefined
|
||||
}
|
||||
return undefined
|
||||
})
|
||||
createDoc.mockImplementation(async (klass: any) => {
|
||||
if (klass === notification.class.DocNotifyContext) {
|
||||
docNotifyContextCreated = true
|
||||
return 'doc-notify-created-id'
|
||||
}
|
||||
return expectedNotificationId
|
||||
})
|
||||
|
||||
await handleScheduledNotification(ctx, workspaceUuid, workSlotMessage, control)
|
||||
|
||||
// 1st createDoc: DocNotifyContext targeting the ToDo, with System modifiedBy.
|
||||
expect(createDoc).toHaveBeenNthCalledWith(
|
||||
1,
|
||||
notification.class.DocNotifyContext,
|
||||
'person-space-1',
|
||||
expect.objectContaining({
|
||||
objectId: 'todo-1',
|
||||
objectClass: 'time:class:ToDo',
|
||||
objectSpace: 'space-1',
|
||||
user: 'person-1'
|
||||
}),
|
||||
undefined,
|
||||
undefined,
|
||||
core.account.System
|
||||
)
|
||||
|
||||
// 2nd createDoc: CommonInboxNotification pointing at the ToDo with deterministic _id.
|
||||
expect(createDoc).toHaveBeenNthCalledWith(
|
||||
2,
|
||||
notification.class.CommonInboxNotification,
|
||||
'person-space-1',
|
||||
expect.objectContaining({
|
||||
user: 'person-1',
|
||||
objectId: 'todo-1',
|
||||
objectClass: 'time:class:ToDo',
|
||||
headerIcon: calendar.icon.Reminder,
|
||||
types: [calendar.ids.ReminderNotification],
|
||||
docNotifyContext: 'doc-notify-created-id'
|
||||
}),
|
||||
expectedNotificationId,
|
||||
undefined,
|
||||
core.account.System
|
||||
)
|
||||
|
||||
// accountClient is NOT consulted on the ToDo path — it's the plain-event path that uses it.
|
||||
expect(findPersonBySocialId).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
// Regression: tracker-issue WorkSlots have `attachedToClass = 'time:class:ProjectToDo'` (a
|
||||
// ToDo subclass created by `IssueToDoFactory`). Strict equality `attachedToClass === ToDo`
|
||||
// would silently take the plain-event branch and route the notification at the WorkSlot
|
||||
// instead of the parent ProjectToDo.
|
||||
it('treats a ProjectToDo-backed WorkSlot as ToDo-backed (subclass via isDerived)', async () => {
|
||||
const projectTodoClass = 'time:class:ProjectToDo' as any
|
||||
findOne.mockImplementation(async (klass: any, query: any) => {
|
||||
if (klass === time.class.WorkSlot) {
|
||||
return {
|
||||
_id: workSlotMessage.eventId,
|
||||
_class: time.class.WorkSlot,
|
||||
space: 'space-1',
|
||||
attachedTo: 'project-todo-1',
|
||||
attachedToClass: projectTodoClass
|
||||
}
|
||||
}
|
||||
if (klass === projectTodoClass) {
|
||||
return {
|
||||
_id: 'project-todo-1',
|
||||
_class: projectTodoClass,
|
||||
space: 'space-1',
|
||||
user: 'employee-1',
|
||||
title: 'Issue ToDo',
|
||||
doneOn: null
|
||||
}
|
||||
}
|
||||
if (klass === contact.mixin.Employee) return { personUuid: 'person-1' }
|
||||
if (klass === contact.class.PersonSpace) return { _id: 'person-space-1' }
|
||||
if (klass === notification.class.CommonInboxNotification) return undefined
|
||||
if (klass === notification.class.DocNotifyContext) return undefined
|
||||
return undefined
|
||||
})
|
||||
createDoc
|
||||
.mockResolvedValueOnce('doc-notify-created-id')
|
||||
.mockResolvedValueOnce(buildReminderNotificationId(workSlotMessage.id))
|
||||
|
||||
await handleScheduledNotification(ctx, workspaceUuid, workSlotMessage, control)
|
||||
|
||||
// hierarchy.isDerived must have been consulted to recognize ProjectToDo as a ToDo.
|
||||
expect(isDerived).toHaveBeenCalledWith(projectTodoClass, time.class.ToDo)
|
||||
|
||||
// The notification points at the ProjectToDo, not at the WorkSlot.
|
||||
expect(createDoc.mock.calls[1][0]).toBe(notification.class.CommonInboxNotification)
|
||||
expect(createDoc.mock.calls[1][2]).toEqual(
|
||||
expect.objectContaining({
|
||||
objectId: 'project-todo-1',
|
||||
objectClass: projectTodoClass
|
||||
})
|
||||
)
|
||||
// accountClient is NOT consulted — the ToDo-backed path doesn't need it.
|
||||
expect(findPersonBySocialId).not.toHaveBeenCalled()
|
||||
})
|
||||
})
|
||||
|
||||
// --------------------------------------------------------------------------------
|
||||
// Plain Event path: notification points at the Event itself.
|
||||
// --------------------------------------------------------------------------------
|
||||
describe('plain Event branch', () => {
|
||||
const expectedNotificationId = buildReminderNotificationId(eventMessage.id)
|
||||
|
||||
it('creates DocNotifyContext + CommonInboxNotification pointing at the Event', async () => {
|
||||
findPersonBySocialId.mockResolvedValue('person-uuid-1')
|
||||
findOne.mockImplementation(async (klass: any, query: any) => {
|
||||
if (klass === eventClassRef) {
|
||||
return {
|
||||
_id: eventMessage.eventId,
|
||||
_class: eventClassRef,
|
||||
space: 'event-space-1',
|
||||
attachedTo: 'some-doc',
|
||||
attachedToClass: 'some:class:Doc',
|
||||
user: 'social:1',
|
||||
title: 'Sprint planning meeting',
|
||||
date: Date.now()
|
||||
}
|
||||
}
|
||||
if (klass === contact.class.Person && query?.personUuid === 'person-uuid-1') {
|
||||
return { _id: 'person-1-doc' }
|
||||
}
|
||||
if (klass === contact.class.PersonSpace) return { _id: 'person-space-1' }
|
||||
if (klass === notification.class.CommonInboxNotification && query?._id === expectedNotificationId) {
|
||||
return undefined
|
||||
}
|
||||
if (klass === notification.class.DocNotifyContext) return undefined
|
||||
return undefined
|
||||
})
|
||||
createDoc
|
||||
.mockResolvedValueOnce('doc-notify-created-id') // DocNotifyContext
|
||||
.mockResolvedValueOnce(expectedNotificationId) // CommonInboxNotification
|
||||
|
||||
await handleScheduledNotification(ctx, workspaceUuid, eventMessage, control)
|
||||
|
||||
expect(findPersonBySocialId).toHaveBeenCalledWith('social:1', true)
|
||||
|
||||
// DocNotifyContext targets the Event itself.
|
||||
expect(createDoc).toHaveBeenNthCalledWith(
|
||||
1,
|
||||
notification.class.DocNotifyContext,
|
||||
'person-space-1',
|
||||
expect.objectContaining({
|
||||
objectId: eventMessage.eventId,
|
||||
objectClass: eventClassRef,
|
||||
objectSpace: 'event-space-1',
|
||||
user: 'person-uuid-1'
|
||||
}),
|
||||
undefined,
|
||||
undefined,
|
||||
core.account.System
|
||||
)
|
||||
|
||||
// Notification targets the Event itself, with the event's title in messageHtml.
|
||||
expect(createDoc).toHaveBeenNthCalledWith(
|
||||
2,
|
||||
notification.class.CommonInboxNotification,
|
||||
'person-space-1',
|
||||
expect.objectContaining({
|
||||
user: 'person-uuid-1',
|
||||
objectId: eventMessage.eventId,
|
||||
objectClass: eventClassRef
|
||||
}),
|
||||
expectedNotificationId,
|
||||
undefined,
|
||||
core.account.System
|
||||
)
|
||||
|
||||
// The notification's title text should be derived from `event.title`, not a ToDo.
|
||||
const notifData = createDoc.mock.calls[1][2]
|
||||
expect(typeof notifData.messageHtml).toBe('string')
|
||||
expect(notifData.messageHtml).toContain('Sprint planning meeting')
|
||||
})
|
||||
|
||||
it('skips when the social id does not resolve to a global person', async () => {
|
||||
findPersonBySocialId.mockResolvedValue(undefined)
|
||||
findOne.mockImplementation(async (klass: any) => {
|
||||
if (klass === eventClassRef) {
|
||||
return {
|
||||
_id: eventMessage.eventId,
|
||||
_class: eventClassRef,
|
||||
space: 'event-space-1',
|
||||
attachedTo: 'some-doc',
|
||||
attachedToClass: 'some:class:Doc',
|
||||
user: 'social:1',
|
||||
title: 'Mystery'
|
||||
}
|
||||
}
|
||||
return undefined
|
||||
})
|
||||
|
||||
await handleScheduledNotification(ctx, workspaceUuid, eventMessage, control)
|
||||
expect(createDoc).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('skips when the event has no `user` (social id) — silently, no error', async () => {
|
||||
findOne.mockImplementation(async (klass: any) => {
|
||||
if (klass === eventClassRef) {
|
||||
return {
|
||||
_id: eventMessage.eventId,
|
||||
_class: eventClassRef,
|
||||
space: 'event-space-1',
|
||||
attachedTo: 'some-doc',
|
||||
attachedToClass: 'some:class:Doc',
|
||||
user: undefined,
|
||||
title: 'Untitled'
|
||||
}
|
||||
}
|
||||
return undefined
|
||||
})
|
||||
|
||||
await handleScheduledNotification(ctx, workspaceUuid, eventMessage, control)
|
||||
|
||||
expect(findPersonBySocialId).not.toHaveBeenCalled()
|
||||
expect(createDoc).not.toHaveBeenCalled()
|
||||
expect(ctx.error).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('skips when the receiver has no PersonSpace', async () => {
|
||||
findPersonBySocialId.mockResolvedValue('person-uuid-1')
|
||||
findOne.mockImplementation(async (klass: any, query: any) => {
|
||||
if (klass === eventClassRef) {
|
||||
return {
|
||||
_id: eventMessage.eventId,
|
||||
_class: eventClassRef,
|
||||
space: 'event-space-1',
|
||||
attachedTo: 'some-doc',
|
||||
attachedToClass: 'some:class:Doc',
|
||||
user: 'social:1',
|
||||
title: 't'
|
||||
}
|
||||
}
|
||||
if (klass === contact.class.Person && query?.personUuid === 'person-uuid-1') return { _id: 'person-1-doc' }
|
||||
return undefined
|
||||
})
|
||||
|
||||
await handleScheduledNotification(ctx, workspaceUuid, eventMessage, control)
|
||||
expect(createDoc).not.toHaveBeenCalled()
|
||||
})
|
||||
})
|
||||
|
||||
// --------------------------------------------------------------------------------
|
||||
// Idempotency + error logging are independent of which branch we took.
|
||||
// --------------------------------------------------------------------------------
|
||||
describe('idempotency and error logging', () => {
|
||||
const expectedNotificationId = buildReminderNotificationId(workSlotMessage.id)
|
||||
|
||||
it('skips creation when reminder was already created (idempotency by _id)', async () => {
|
||||
findOne.mockImplementation(async (klass: any, query: any) => {
|
||||
if (klass === time.class.WorkSlot) {
|
||||
return {
|
||||
_id: workSlotMessage.eventId,
|
||||
_class: time.class.WorkSlot,
|
||||
space: 'space-1',
|
||||
attachedTo: 'todo-1',
|
||||
attachedToClass: time.class.ToDo
|
||||
}
|
||||
}
|
||||
if (klass === time.class.ToDo) {
|
||||
return {
|
||||
_id: 'todo-1',
|
||||
_class: 'time:class:ToDo',
|
||||
space: 'space-1',
|
||||
user: 'employee-1',
|
||||
title: 'Todo title'
|
||||
}
|
||||
}
|
||||
if (klass === contact.mixin.Employee) return { personUuid: 'person-1' }
|
||||
if (klass === contact.class.PersonSpace) return { _id: 'person-space-1' }
|
||||
if (klass === notification.class.CommonInboxNotification && query?._id === expectedNotificationId) {
|
||||
return { _id: expectedNotificationId }
|
||||
}
|
||||
return undefined
|
||||
})
|
||||
|
||||
await handleScheduledNotification(ctx, workspaceUuid, workSlotMessage, control)
|
||||
|
||||
expect(createDoc).not.toHaveBeenCalled()
|
||||
expect(ctx.info).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('reuses an existing DocNotifyContext without creating a new one', async () => {
|
||||
findOne.mockImplementation(async (klass: any, query: any) => {
|
||||
if (klass === time.class.WorkSlot) {
|
||||
return {
|
||||
_id: workSlotMessage.eventId,
|
||||
_class: time.class.WorkSlot,
|
||||
space: 'space-1',
|
||||
attachedTo: 'todo-1',
|
||||
attachedToClass: time.class.ToDo
|
||||
}
|
||||
}
|
||||
if (klass === time.class.ToDo) {
|
||||
return { _id: 'todo-1', _class: 'time:class:ToDo', space: 'space-1', user: 'employee-1', title: 't' }
|
||||
}
|
||||
if (klass === contact.mixin.Employee) return { personUuid: 'person-1' }
|
||||
if (klass === contact.class.PersonSpace) return { _id: 'person-space-1' }
|
||||
if (klass === notification.class.CommonInboxNotification && query?._id === expectedNotificationId) {
|
||||
return undefined
|
||||
}
|
||||
if (klass === notification.class.DocNotifyContext) return { _id: 'existing-doc-notify' }
|
||||
return undefined
|
||||
})
|
||||
createDoc.mockResolvedValue(expectedNotificationId)
|
||||
|
||||
await handleScheduledNotification(ctx, workspaceUuid, workSlotMessage, control)
|
||||
|
||||
expect(createDoc).toHaveBeenCalledTimes(1)
|
||||
expect(createDoc).toHaveBeenCalledWith(
|
||||
notification.class.CommonInboxNotification,
|
||||
'person-space-1',
|
||||
expect.objectContaining({ docNotifyContext: 'existing-doc-notify' }),
|
||||
expectedNotificationId,
|
||||
undefined,
|
||||
core.account.System
|
||||
)
|
||||
})
|
||||
|
||||
it('logs a detailed error and rethrows when CommonInboxNotification createDoc fails with Bad Request', async () => {
|
||||
findOne.mockImplementation(async (klass: any, query: any) => {
|
||||
if (klass === time.class.WorkSlot) {
|
||||
return {
|
||||
_id: workSlotMessage.eventId,
|
||||
_class: time.class.WorkSlot,
|
||||
space: 'space-1',
|
||||
attachedTo: 'todo-1',
|
||||
attachedToClass: time.class.ToDo
|
||||
}
|
||||
}
|
||||
if (klass === time.class.ToDo) {
|
||||
return { _id: 'todo-1', _class: 'time:class:ToDo', space: 'space-1', user: 'employee-1', title: 't' }
|
||||
}
|
||||
if (klass === contact.mixin.Employee) return { personUuid: 'person-1' }
|
||||
if (klass === contact.class.PersonSpace) return { _id: 'person-space-1' }
|
||||
if (klass === notification.class.CommonInboxNotification && query?._id === expectedNotificationId) {
|
||||
return undefined
|
||||
}
|
||||
if (klass === notification.class.DocNotifyContext) return { _id: 'existing-doc-notify' }
|
||||
return undefined
|
||||
})
|
||||
const badRequest = new PlatformError(
|
||||
new Status(Severity.ERROR, 'platform:status:UnknownError' as any, { message: 'Bad Request' })
|
||||
)
|
||||
createDoc.mockRejectedValueOnce(badRequest)
|
||||
|
||||
await expect(handleScheduledNotification(ctx, workspaceUuid, workSlotMessage, control)).rejects.toBe(badRequest)
|
||||
|
||||
expect(ctx.error).toHaveBeenCalledWith(
|
||||
'Failed to create CommonInboxNotification for event reminder',
|
||||
expect.objectContaining({
|
||||
err: badRequest,
|
||||
timerId: workSlotMessage.id,
|
||||
eventId: workSlotMessage.eventId,
|
||||
eventClass: workSlotMessage.eventClass,
|
||||
notificationId: expectedNotificationId,
|
||||
spaceId: 'person-space-1',
|
||||
user: 'person-1'
|
||||
})
|
||||
)
|
||||
})
|
||||
|
||||
it('logs a detailed error and rethrows when DocNotifyContext createDoc fails', async () => {
|
||||
findOne.mockImplementation(async (klass: any) => {
|
||||
if (klass === time.class.WorkSlot) {
|
||||
return {
|
||||
_id: workSlotMessage.eventId,
|
||||
_class: time.class.WorkSlot,
|
||||
space: 'space-1',
|
||||
attachedTo: 'todo-1',
|
||||
attachedToClass: time.class.ToDo
|
||||
}
|
||||
}
|
||||
if (klass === time.class.ToDo) {
|
||||
return { _id: 'todo-1', _class: 'time:class:ToDo', space: 'space-1', user: 'employee-1', title: 't' }
|
||||
}
|
||||
if (klass === contact.mixin.Employee) return { personUuid: 'person-1' }
|
||||
if (klass === contact.class.PersonSpace) return { _id: 'person-space-1' }
|
||||
if (klass === notification.class.CommonInboxNotification) return undefined
|
||||
if (klass === notification.class.DocNotifyContext) return undefined
|
||||
return undefined
|
||||
})
|
||||
const badRequest = new PlatformError(
|
||||
new Status(Severity.ERROR, 'platform:status:UnknownError' as any, { message: 'Bad Request' })
|
||||
)
|
||||
createDoc.mockRejectedValueOnce(badRequest)
|
||||
|
||||
await expect(handleScheduledNotification(ctx, workspaceUuid, workSlotMessage, control)).rejects.toBe(badRequest)
|
||||
|
||||
expect(ctx.error).toHaveBeenCalledWith(
|
||||
'Failed to create DocNotifyContext for event reminder',
|
||||
expect.objectContaining({
|
||||
err: badRequest,
|
||||
timerId: workSlotMessage.id,
|
||||
eventId: workSlotMessage.eventId,
|
||||
eventClass: workSlotMessage.eventClass,
|
||||
spaceId: 'person-space-1',
|
||||
user: 'person-1'
|
||||
})
|
||||
)
|
||||
expect(createDoc).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
|
||||
it('builds a stable, distinct notification _id per timer', () => {
|
||||
const a = buildReminderNotificationId('timer-A')
|
||||
const b = buildReminderNotificationId('timer-B')
|
||||
expect(a).not.toBe(b)
|
||||
expect(buildReminderNotificationId('timer-A')).toBe(a)
|
||||
expect(a.startsWith('eventReminderInbox:')).toBe(true)
|
||||
})
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,78 @@
|
||||
//
|
||||
// Copyright © 2026 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 { getClient as getAccountClient, type AccountClient } from '@hcengineering/account-client'
|
||||
import { createRestTxOperations } from '@hcengineering/api-client'
|
||||
import core, { PersonId, systemAccountUuid, type TxOperations, type WorkspaceUuid } from '@hcengineering/core'
|
||||
import { generateToken } from '@hcengineering/server-token'
|
||||
import {
|
||||
clearInFlightClientCreation,
|
||||
getCacheKey,
|
||||
getCachedClient,
|
||||
getInFlightClientCreation,
|
||||
setCachedClient,
|
||||
setInFlightClientCreation
|
||||
} from './clientCache'
|
||||
import config from './config'
|
||||
|
||||
export interface ClientBundle {
|
||||
client: TxOperations
|
||||
accountClient: AccountClient
|
||||
}
|
||||
|
||||
export async function getClient (
|
||||
workspaceUuid: WorkspaceUuid,
|
||||
socialId?: PersonId,
|
||||
serviceTag: string = config.ServiceId
|
||||
): Promise<ClientBundle> {
|
||||
const cached = getCachedClient(workspaceUuid, socialId, serviceTag)
|
||||
if (cached !== undefined) return cached
|
||||
|
||||
const cacheKey = getCacheKey(workspaceUuid, socialId, serviceTag)
|
||||
const inFlight = getInFlightClientCreation(cacheKey)
|
||||
if (inFlight !== undefined) return await inFlight
|
||||
|
||||
const creation = (async () => {
|
||||
const token = generateToken(systemAccountUuid, workspaceUuid, { service: serviceTag })
|
||||
let accountClient = getAccountClient(config.AccountsUrl, token)
|
||||
|
||||
// If we want the notification author to be a specific user, we can obtain a workspace token for that person.
|
||||
if (socialId !== undefined && socialId !== core.account.System) {
|
||||
const personUuid = await accountClient.findPersonBySocialId(socialId, true)
|
||||
if (personUuid === undefined) {
|
||||
throw new Error(`Global person not found for social-id ${socialId}`)
|
||||
}
|
||||
const token = generateToken(personUuid, workspaceUuid, { service: serviceTag })
|
||||
accountClient = getAccountClient(config.AccountsUrl, token)
|
||||
}
|
||||
|
||||
const wsInfo = await accountClient.getLoginInfoByToken()
|
||||
if (wsInfo == null || !('endpoint' in wsInfo)) {
|
||||
throw new Error('Invalid login info')
|
||||
}
|
||||
const transactorUrl = wsInfo.endpoint.replace('ws://', 'http://').replace('wss://', 'https://')
|
||||
const client = await createRestTxOperations(transactorUrl, wsInfo.workspace, wsInfo.token)
|
||||
const bundle = { client, accountClient }
|
||||
setCachedClient(workspaceUuid, socialId, serviceTag, bundle)
|
||||
return bundle
|
||||
})()
|
||||
|
||||
setInFlightClientCreation(cacheKey, creation)
|
||||
try {
|
||||
return await creation
|
||||
} finally {
|
||||
clearInFlightClientCreation(cacheKey)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,72 @@
|
||||
//
|
||||
// Copyright © 2026 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 PersonId, type WorkspaceUuid } from '@hcengineering/core'
|
||||
import type { ClientBundle } from './client'
|
||||
import config from './config'
|
||||
|
||||
interface CacheEntry {
|
||||
value: ClientBundle
|
||||
expiresAt: number
|
||||
}
|
||||
|
||||
const clientCache = new Map<string, CacheEntry>()
|
||||
const inFlightClientCreations = new Map<string, Promise<ClientBundle>>()
|
||||
|
||||
export function getCacheKey (workspaceUuid: WorkspaceUuid, socialId: PersonId | undefined, serviceTag: string): string {
|
||||
return `${workspaceUuid}:${socialId ?? 'system'}:${serviceTag}`
|
||||
}
|
||||
|
||||
export function getCachedClient (
|
||||
workspaceUuid: WorkspaceUuid,
|
||||
socialId: PersonId | undefined,
|
||||
serviceTag: string
|
||||
): ClientBundle | undefined {
|
||||
const key = getCacheKey(workspaceUuid, socialId, serviceTag)
|
||||
const entry = clientCache.get(key)
|
||||
if (entry === undefined) return undefined
|
||||
if (entry.expiresAt <= Date.now()) {
|
||||
clientCache.delete(key)
|
||||
return undefined
|
||||
}
|
||||
return entry.value
|
||||
}
|
||||
|
||||
export function setCachedClient (
|
||||
workspaceUuid: WorkspaceUuid,
|
||||
socialId: PersonId | undefined,
|
||||
serviceTag: string,
|
||||
value: ClientBundle
|
||||
): void {
|
||||
const key = getCacheKey(workspaceUuid, socialId, serviceTag)
|
||||
clientCache.set(key, { value, expiresAt: Date.now() + config.ClientCacheTtlMs })
|
||||
}
|
||||
|
||||
export function getInFlightClientCreation (key: string): Promise<ClientBundle> | undefined {
|
||||
return inFlightClientCreations.get(key)
|
||||
}
|
||||
|
||||
export function setInFlightClientCreation (key: string, creation: Promise<ClientBundle>): void {
|
||||
inFlightClientCreations.set(key, creation)
|
||||
}
|
||||
|
||||
export function clearInFlightClientCreation (key: string): void {
|
||||
inFlightClientCreations.delete(key)
|
||||
}
|
||||
|
||||
export function clearClientCachesForTests (): void {
|
||||
clientCache.clear()
|
||||
inFlightClientCreations.clear()
|
||||
}
|
||||
@@ -0,0 +1,38 @@
|
||||
//
|
||||
// Copyright © 2026 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()
|
||||
|
||||
interface Config {
|
||||
ServiceId: string
|
||||
Secret: string
|
||||
AccountsUrl: string
|
||||
QueueRegion: string
|
||||
LogLevel: 'info' | 'debug'
|
||||
ClientCacheTtlMs: number
|
||||
}
|
||||
|
||||
const config: Config = {
|
||||
ServiceId: process.env.SERVICE_ID ?? 'events-processor',
|
||||
Secret: process.env.SECRET ?? 'secret',
|
||||
AccountsUrl: process.env.ACCOUNTS_URL ?? 'http://localhost:3000',
|
||||
QueueRegion: process.env.QUEUE_REGION ?? 'localhost',
|
||||
LogLevel: process.env.LOG_LEVEL === 'debug' ? 'debug' : 'info',
|
||||
ClientCacheTtlMs: Number(process.env.CLIENT_CACHE_TTL_MS ?? 60_000)
|
||||
}
|
||||
|
||||
export default config
|
||||
@@ -0,0 +1,98 @@
|
||||
//
|
||||
// Copyright © 2026 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 { Analytics } from '@hcengineering/analytics'
|
||||
import { configureAnalytics, createOpenTelemetryMetricsContext, SplitLogger } from '@hcengineering/analytics-service'
|
||||
import { newMetrics } from '@hcengineering/core'
|
||||
import { getPlatformQueue } from '@hcengineering/kafka'
|
||||
import { setMetadata } from '@hcengineering/platform'
|
||||
import serverClient from '@hcengineering/server-client'
|
||||
import { initStatisticsContext, type ConsumerControl } from '@hcengineering/server-core'
|
||||
import serverToken from '@hcengineering/server-token'
|
||||
import { join } from 'path'
|
||||
import config from './config'
|
||||
import type { ScheduledNotificationMessage } from './types'
|
||||
import { handleScheduledNotification } from './worker'
|
||||
|
||||
const scheduledNotificationTopic = 'scheduledNotification'
|
||||
const isDebugLoggingEnabled = config.LogLevel === 'debug'
|
||||
const serviceVersion = process.env.VERSION ?? '0.7.0'
|
||||
|
||||
async function main (): Promise<void> {
|
||||
configureAnalytics(config.ServiceId, serviceVersion)
|
||||
const ctx = initStatisticsContext(config.ServiceId, {
|
||||
factory: () =>
|
||||
createOpenTelemetryMetricsContext(
|
||||
config.ServiceId,
|
||||
{},
|
||||
{},
|
||||
newMetrics(),
|
||||
new SplitLogger(config.ServiceId, {
|
||||
root: join(process.cwd(), 'logs'),
|
||||
enableConsole: (process.env.ENABLE_CONSOLE ?? 'true') === 'true'
|
||||
})
|
||||
)
|
||||
})
|
||||
|
||||
Analytics.setTag('application', config.ServiceId)
|
||||
setMetadata(serverToken.metadata.Secret, config.Secret)
|
||||
setMetadata(serverToken.metadata.Service, config.ServiceId)
|
||||
setMetadata(serverClient.metadata.Endpoint, config.AccountsUrl)
|
||||
|
||||
const queue = getPlatformQueue(config.ServiceId, config.QueueRegion)
|
||||
|
||||
const consumer = queue.createConsumer<ScheduledNotificationMessage>(
|
||||
ctx,
|
||||
scheduledNotificationTopic,
|
||||
queue.getClientId(),
|
||||
async (ctx, message, control: ConsumerControl) => {
|
||||
if (isDebugLoggingEnabled) {
|
||||
ctx.info('Received scheduled notification event', {
|
||||
topic: scheduledNotificationTopic,
|
||||
workspace: message.workspace,
|
||||
kind: message.value?.kind,
|
||||
id: message.value?.id
|
||||
})
|
||||
}
|
||||
await handleScheduledNotification(ctx, message.workspace, message.value, control)
|
||||
}
|
||||
)
|
||||
|
||||
ctx.info(`Started events processor: version ${serviceVersion}`, {
|
||||
serviceId: config.ServiceId,
|
||||
version: serviceVersion,
|
||||
topic: scheduledNotificationTopic,
|
||||
queueRegion: config.QueueRegion,
|
||||
logLevel: config.LogLevel
|
||||
})
|
||||
|
||||
const shutdown = (): void => {
|
||||
void Promise.all([consumer.close()]).then(() => process.exit())
|
||||
}
|
||||
|
||||
process.once('SIGINT', shutdown)
|
||||
process.once('SIGTERM', shutdown)
|
||||
process.on('uncaughtException', (error: any) => {
|
||||
ctx.error('Uncaught exception', { error })
|
||||
})
|
||||
process.on('unhandledRejection', (error: any) => {
|
||||
ctx.error('Unhandled rejection', { error })
|
||||
})
|
||||
}
|
||||
|
||||
void main().catch((err) => {
|
||||
// eslint-disable-next-line no-console
|
||||
console.error(err)
|
||||
})
|
||||
@@ -0,0 +1,31 @@
|
||||
//
|
||||
// Copyright © 2026 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 { Class, Doc, Ref } from '@hcengineering/core'
|
||||
|
||||
/**
|
||||
* Scheduled notification fired by the time-machine consumer.
|
||||
*
|
||||
* `kind === 'eventReminder'` covers any `calendar.Event` subclass — including `WorkSlot`,
|
||||
* which is handled with a small ToDo-specific branch in the worker.
|
||||
*/
|
||||
export interface ScheduledNotificationMessage {
|
||||
kind: 'eventReminder'
|
||||
id: string
|
||||
eventId: Ref<Doc>
|
||||
eventClass: Ref<Class<Doc>>
|
||||
shiftMs: number
|
||||
targetDate: number
|
||||
}
|
||||
@@ -0,0 +1,233 @@
|
||||
//
|
||||
// Copyright © 2026 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 calendar from '@hcengineering/calendar'
|
||||
import contact, { type Person, type PersonSpace } from '@hcengineering/contact'
|
||||
import core, {
|
||||
type AccountUuid,
|
||||
type Class,
|
||||
type Doc,
|
||||
type Hierarchy,
|
||||
type MeasureContext,
|
||||
type PersonId,
|
||||
type Ref,
|
||||
type Space,
|
||||
type WorkspaceUuid
|
||||
} from '@hcengineering/core'
|
||||
import type { ConsumerControl } from '@hcengineering/server-core'
|
||||
import notification, { type CommonInboxNotification, type DocNotifyContext } from '@hcengineering/notification'
|
||||
import { jsonToMarkup, nodeDoc, nodeParagraph, nodeText } from '@hcengineering/text-core'
|
||||
import time, { type ToDo } from '@hcengineering/time'
|
||||
import { getClient, type ClientBundle } from './client'
|
||||
import type { ScheduledNotificationMessage } from './types'
|
||||
|
||||
interface MinimalEvent extends Doc {
|
||||
attachedTo?: Ref<Doc>
|
||||
attachedToClass?: Ref<Class<Doc>>
|
||||
user?: PersonId
|
||||
title?: string
|
||||
}
|
||||
|
||||
export function buildReminderNotificationId (timerId: string): Ref<CommonInboxNotification> {
|
||||
return `eventReminderInbox:${timerId}` as Ref<CommonInboxNotification>
|
||||
}
|
||||
|
||||
interface ReminderTarget {
|
||||
objectId: Ref<any>
|
||||
objectClass: Ref<Class<any>>
|
||||
objectSpace: Ref<Space>
|
||||
titleText: string
|
||||
receiverAccount: AccountUuid
|
||||
receiverSpace: Ref<PersonSpace>
|
||||
}
|
||||
|
||||
export async function handleScheduledNotification (
|
||||
ctx: MeasureContext,
|
||||
workspaceUuid: WorkspaceUuid,
|
||||
msg: ScheduledNotificationMessage,
|
||||
control: ConsumerControl
|
||||
): Promise<void> {
|
||||
if (msg.kind !== 'eventReminder') return
|
||||
|
||||
await control.heartbeat()
|
||||
const bundle = await getClient(workspaceUuid)
|
||||
const { client } = bundle
|
||||
await control.heartbeat()
|
||||
|
||||
const event = (await client.findOne(msg.eventClass, { _id: msg.eventId })) as MinimalEvent | undefined
|
||||
if (event === undefined) return
|
||||
await control.heartbeat()
|
||||
|
||||
const hierarchy = client.getHierarchy()
|
||||
const target = await resolveReminderTarget(bundle, event, hierarchy)
|
||||
if (target === undefined) return
|
||||
await control.heartbeat()
|
||||
|
||||
// Idempotency: short-circuit if a notification for this exact timer was already created.
|
||||
const notificationId = buildReminderNotificationId(msg.id)
|
||||
const existing = await client.findOne(
|
||||
notification.class.CommonInboxNotification,
|
||||
{ _id: notificationId },
|
||||
{ projection: { _id: 1 } }
|
||||
)
|
||||
if (existing !== undefined) return
|
||||
|
||||
// Ensure doc notify context exists.
|
||||
let docNotifyContext: Pick<DocNotifyContext, '_id'> | undefined = await client.findOne(
|
||||
notification.class.DocNotifyContext,
|
||||
{ objectId: target.objectId, user: target.receiverAccount },
|
||||
{ projection: { _id: 1 } }
|
||||
)
|
||||
if (docNotifyContext === undefined) {
|
||||
try {
|
||||
const id = await client.createDoc(
|
||||
notification.class.DocNotifyContext,
|
||||
target.receiverSpace,
|
||||
{
|
||||
objectId: target.objectId,
|
||||
objectClass: target.objectClass,
|
||||
objectSpace: target.objectSpace,
|
||||
user: target.receiverAccount,
|
||||
isPinned: false,
|
||||
hidden: false
|
||||
},
|
||||
undefined,
|
||||
undefined,
|
||||
// System tokens may not have a populated `socialIds[0]`; if we let the rest tx client default
|
||||
// to that, `Tx.modifiedBy` becomes `undefined` and the transactor rejects the request with
|
||||
// HTTP 400 "Bad Request" out of `NormalizeTxMiddleware.parseBaseTx`. Always pin to a known PersonId.
|
||||
core.account.System
|
||||
)
|
||||
docNotifyContext = { _id: id }
|
||||
} catch (err) {
|
||||
ctx.error('Failed to create DocNotifyContext for event reminder', {
|
||||
err,
|
||||
timerId: msg.id,
|
||||
eventId: msg.eventId,
|
||||
eventClass: msg.eventClass,
|
||||
spaceId: target.receiverSpace,
|
||||
user: target.receiverAccount
|
||||
})
|
||||
throw err
|
||||
}
|
||||
}
|
||||
|
||||
await control.heartbeat()
|
||||
try {
|
||||
await client.createDoc(
|
||||
notification.class.CommonInboxNotification,
|
||||
target.receiverSpace,
|
||||
{
|
||||
user: target.receiverAccount,
|
||||
objectId: target.objectId,
|
||||
objectClass: target.objectClass,
|
||||
headerIcon: calendar.icon.Reminder,
|
||||
header: calendar.string.Reminder,
|
||||
message: calendar.string.Reminder,
|
||||
messageHtml: jsonToMarkup(nodeDoc(nodeParagraph(nodeText(target.titleText)))),
|
||||
types: [calendar.ids.ReminderNotification],
|
||||
isViewed: false,
|
||||
archived: false,
|
||||
docNotifyContext: docNotifyContext._id
|
||||
},
|
||||
notificationId,
|
||||
undefined,
|
||||
core.account.System
|
||||
)
|
||||
} catch (err) {
|
||||
ctx.error('Failed to create CommonInboxNotification for event reminder', {
|
||||
err,
|
||||
timerId: msg.id,
|
||||
eventId: msg.eventId,
|
||||
eventClass: msg.eventClass,
|
||||
notificationId,
|
||||
spaceId: target.receiverSpace,
|
||||
user: target.receiverAccount
|
||||
})
|
||||
throw err
|
||||
}
|
||||
|
||||
ctx.info('Scheduled notification created', {
|
||||
kind: msg.kind,
|
||||
id: msg.id,
|
||||
eventId: msg.eventId,
|
||||
eventClass: msg.eventClass,
|
||||
notificationId,
|
||||
user: target.receiverAccount,
|
||||
spaceId: target.receiverSpace
|
||||
})
|
||||
}
|
||||
|
||||
// Resolves where a reminder fires:
|
||||
// - For Events whose `attachedToClass` is `time:class:ToDo` (or a subclass like `ProjectToDo`),
|
||||
// the notification points at the parent ToDo and is suppressed when the ToDo is already done.
|
||||
// - For any other Event the notification points at the Event itself.
|
||||
async function resolveReminderTarget (
|
||||
bundle: ClientBundle,
|
||||
event: MinimalEvent,
|
||||
hierarchy: Hierarchy
|
||||
): Promise<ReminderTarget | undefined> {
|
||||
const { client, accountClient } = bundle
|
||||
|
||||
const isToDoBacked = event.attachedToClass != null && hierarchy.isDerived(event.attachedToClass, time.class.ToDo)
|
||||
if (isToDoBacked && event.attachedTo != null && event.attachedToClass != null) {
|
||||
const todo = (await client.findOne(event.attachedToClass as Ref<Class<ToDo>>, {
|
||||
_id: event.attachedTo as Ref<ToDo>
|
||||
})) as ToDo | undefined
|
||||
if (todo === undefined) return undefined
|
||||
if (todo.doneOn != null) return undefined
|
||||
|
||||
const employee = await client.findOne(contact.mixin.Employee, { _id: todo.user, active: true })
|
||||
if (employee?.personUuid == null) return undefined
|
||||
|
||||
const space = await client.findOne(contact.class.PersonSpace, { person: todo.user }, { projection: { _id: 1 } })
|
||||
if (space === undefined) return undefined
|
||||
|
||||
return {
|
||||
objectId: todo._id,
|
||||
objectClass: todo._class,
|
||||
objectSpace: todo.space,
|
||||
titleText: todo.title ?? '',
|
||||
receiverAccount: employee.personUuid,
|
||||
receiverSpace: space._id
|
||||
}
|
||||
}
|
||||
|
||||
// Plain calendar event: receiver is identified by `event.user` (a PersonId / social id).
|
||||
const receiverSocialId = event.user
|
||||
if (receiverSocialId == null) return undefined
|
||||
|
||||
const personUuid = await accountClient.findPersonBySocialId(receiverSocialId, true)
|
||||
if (personUuid == null) return undefined
|
||||
|
||||
const person = await client.findOne(contact.class.Person, { personUuid }, { projection: { _id: 1 } })
|
||||
if (person === undefined) return undefined
|
||||
|
||||
const space = await client.findOne(
|
||||
contact.class.PersonSpace,
|
||||
{ person: person._id as Ref<Person> },
|
||||
{ projection: { _id: 1 } }
|
||||
)
|
||||
if (space === undefined) return undefined
|
||||
|
||||
return {
|
||||
objectId: event._id,
|
||||
objectClass: event._class,
|
||||
objectSpace: event.space,
|
||||
titleText: event.title ?? '',
|
||||
receiverAccount: personUuid as AccountUuid,
|
||||
receiverSpace: space._id
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,12 @@
|
||||
{
|
||||
"extends": "./node_modules/@hcengineering/platform-rig/profiles/default/tsconfig.json",
|
||||
"compilerOptions": {
|
||||
"rootDir": "./src",
|
||||
"outDir": "./lib",
|
||||
"declarationDir": "./types",
|
||||
"tsBuildInfoFile": ".build/build.tsbuildinfo"
|
||||
},
|
||||
"include": ["src/**/*"],
|
||||
"exclude": ["node_modules", "lib", "dist", "types", "bundle"]
|
||||
}
|
||||
|
||||
@@ -33,7 +33,8 @@ When a timer expires, the service relays the exact `data` payload to the target
|
||||
| `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. |
|
||||
| `QUEUE_REGION` | (empty) | Kafka topic prefix; must match transactor `REGION` / other services `QUEUE_REGION`. |
|
||||
| `LOG_LEVEL` | `info` | Set to `debug` for verbose diagnostic logs (`ctx.debug`) on TimeMachine consumes and poll batches. |
|
||||
|
||||
## Database Schema
|
||||
|
||||
|
||||
@@ -21,13 +21,15 @@ export interface Config {
|
||||
PollInterval: number
|
||||
QueueRegion: string
|
||||
QueueConfig: string
|
||||
LogLevel: 'info' | 'debug'
|
||||
}
|
||||
|
||||
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 ?? '',
|
||||
QueueConfig: process.env.QUEUE_CONFIG ?? ''
|
||||
QueueConfig: process.env.QUEUE_CONFIG ?? '',
|
||||
LogLevel: process.env.LOG_LEVEL === 'debug' ? 'debug' : 'info'
|
||||
}
|
||||
|
||||
export default config
|
||||
|
||||
@@ -13,7 +13,7 @@
|
||||
// limitations under the License.
|
||||
//
|
||||
|
||||
import { MeasureMetricsContext } from '@hcengineering/core'
|
||||
import { MeasureMetricsContext, newMetrics } from '@hcengineering/core'
|
||||
import { getPlatformQueue } from '@hcengineering/kafka'
|
||||
import { QueueTopic } from '@hcengineering/server-core'
|
||||
import { TimeMachineMessage } from '@hcengineering/server-process'
|
||||
@@ -25,13 +25,29 @@ export async function runWorker (): Promise<void> {
|
||||
const SERVICE_NAME = 'time-machine'
|
||||
const db = await TimeMachineDB.init(config.DbUrl)
|
||||
|
||||
const ctx = new MeasureMetricsContext(SERVICE_NAME, {})
|
||||
const ctx = new MeasureMetricsContext(
|
||||
SERVICE_NAME,
|
||||
{},
|
||||
{},
|
||||
newMetrics(),
|
||||
undefined,
|
||||
undefined,
|
||||
undefined,
|
||||
config.LogLevel
|
||||
)
|
||||
const queue = getPlatformQueue(SERVICE_NAME, config.QueueRegion)
|
||||
|
||||
// 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) {
|
||||
ctx.debug('TimeMachine consume schedule', {
|
||||
workspace: msg.workspace,
|
||||
id,
|
||||
targetDate,
|
||||
targetDateIso: new Date(targetDate).toISOString(),
|
||||
outTopic: topic
|
||||
})
|
||||
await db.upsertEvent({
|
||||
id,
|
||||
workspace: msg.workspace,
|
||||
@@ -40,6 +56,7 @@ export async function runWorker (): Promise<void> {
|
||||
data
|
||||
})
|
||||
} else if (type === 'cancel') {
|
||||
ctx.debug('TimeMachine consume cancel', { workspace: msg.workspace, idPattern: id })
|
||||
await db.removeEvents(msg.workspace, id)
|
||||
}
|
||||
})
|
||||
@@ -49,13 +66,24 @@ export async function runWorker (): Promise<void> {
|
||||
try {
|
||||
const expiredEvents = await db.getExpiredEvents()
|
||||
if (expiredEvents.length > 0) {
|
||||
ctx.debug('TimeMachine poll expired', {
|
||||
count: expiredEvents.length,
|
||||
ids: expiredEvents.map((e) => e.id)
|
||||
})
|
||||
for (const event of expiredEvents) {
|
||||
ctx.debug('TimeMachine relay', {
|
||||
workspace: event.workspace,
|
||||
id: event.id,
|
||||
outTopic: event.topic,
|
||||
targetDate: event.target_date,
|
||||
targetDateIso: new Date(event.target_date).toISOString()
|
||||
})
|
||||
await SendTimeEvent(ctx, event.workspace, event.topic, event.data)
|
||||
}
|
||||
await db.deleteEvents(expiredEvents)
|
||||
}
|
||||
} catch (err) {
|
||||
ctx.error('Error in Time Machine polling loop:')
|
||||
ctx.error('Error in Time Machine polling loop', { err })
|
||||
} finally {
|
||||
setTimeout(() => {
|
||||
void poll()
|
||||
|
||||
Reference in New Issue
Block a user