diff --git a/.vscode/launch.json b/.vscode/launch.json index 15bc2982e1..542eb32e20 100644 --- a/.vscode/launch.json +++ b/.vscode/launch.json @@ -826,7 +826,9 @@ "request": "launch", "args": ["src/index.ts"], "env": { + "ACCOUNTS_URL": "http://huly.local:3000", "SECRET": "secret", + "REGION": "cockroach", "QUEUE_CONFIG": "localhost:19092" }, "runtimeArgs": ["--nolazy", "-r", "ts-node/register"], diff --git a/dev/docker-compose.yaml b/dev/docker-compose.yaml index 65844a49a5..8f1324d40c 100644 --- a/dev/docker-compose.yaml +++ b/dev/docker-compose.yaml @@ -9,8 +9,10 @@ services: - STREAM_INSECURE=true - STREAM_SERVER_SECRET=secret - STREAM_MAX_PARALLEL_SCALING_COUNT=6 + - STREAM_LOG_LEVEL=debug - AWS_ACCESS_KEY_ID=minioadmin - AWS_SECRET_ACCESS_KEY=minioadmin + - STREAM_REGION=cockroach - STREAM_QUEUE_CONFIG=${QUEUE_CONFIG} ports: - 1080:1080 @@ -21,7 +23,9 @@ services: - 'huly.local:host-gateway' container_name: media environment: + - ACCOUNTS_URL=http://huly.local:3000 - SECRET=secret + - REGION=cockroach - QUEUE_CONFIG=${QUEUE_CONFIG} restart: unless-stopped cockroach: @@ -333,6 +337,7 @@ services: - 4005:4005 environment: - SECRET=secret + - REGION=cockroach - QUEUE_CONFIG=${QUEUE_CONFIG} - STORAGE_CONFIG=${STORAGE_CONFIG} - STATS_URL=http://huly.local:4900 @@ -350,6 +355,7 @@ services: environment: - SECRET=secret - MINIO_ENDPOINT=minio + - REGION=cockroach - QUEUE_CONFIG=${QUEUE_CONFIG} - MINIO_ACCESS_KEY=minioadmin - ACCOUNTS_URL=http://huly.local:3000 @@ -432,6 +438,7 @@ services: - STATS_URL=http://huly.local:4900 - DB_URL=${DB_CR_URL} - BUCKETS=blobs,eu|http://minio:9000?accessKey=minioadmin&secretKey=minioadmin + - REGION=cockroach - QUEUE_CONFIG=${QUEUE_CONFIG} restart: unless-stopped hulykvs: diff --git a/pods/media/package.json b/pods/media/package.json index dce179999e..b5bbf882cb 100644 --- a/pods/media/package.json +++ b/pods/media/package.json @@ -57,9 +57,15 @@ "@hcengineering/core": "^0.6.32", "@hcengineering/kafka": "^0.6.0", "@hcengineering/platform": "^0.6.11", + "@hcengineering/api-client": "^0.6.0", + "@hcengineering/server-client": "^0.6.0", "@hcengineering/server-core": "^0.6.1", "@hcengineering/server-storage": "^0.6.0", "@hcengineering/server-token": "^0.6.11", + "@hcengineering/attachment": "^0.6.14", + "@hcengineering/drive": "^0.6.0", + "@hcengineering/communication-types": "^0.1.0", + "@hcengineering/communication-sdk-types": "^0.1.0", "dotenv": "~16.0.0", "kafkajs": "^2.2.4" } diff --git a/pods/media/src/client.ts b/pods/media/src/client.ts new file mode 100644 index 0000000000..4df356949a --- /dev/null +++ b/pods/media/src/client.ts @@ -0,0 +1,186 @@ +// +// Copyright © 2025 Hardcore Engineering Inc. +// +// Licensed under the Eclipse Public License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. You may +// obtain a copy of the License at https://www.eclipse.org/legal/epl-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// +// See the License for the specific language governing permissions and +// limitations under the License. +// + +import attachment, { type Attachment } from '@hcengineering/attachment' +import drive, { FileVersion } from '@hcengineering/drive' +import core, { + BlobMetadata, + Class, + Client, + Doc, + DocumentQuery, + DomainParams, + DomainRequestOptions, + DomainResult, + FindOptions, + FindResult, + Hierarchy, + MeasureContext, + ModelDb, + OperationDomain, + Ref, + SearchOptions, + SearchQuery, + SearchResult, + systemAccountUuid, + Tx, + TxOperations, + TxResult, + WithLookup, + WorkspaceUuid +} from '@hcengineering/core' +import { type RestClient, createRestClient } from '@hcengineering/api-client' +import { getTransactorEndpoint } from '@hcengineering/server-client' +import { generateToken } from '@hcengineering/server-token' + +import { BlobSourceType, type VideoTranscodeResult } from './types' + +async function getClient (workspace: WorkspaceUuid, token: string): Promise { + const endpoint = await getTransactorEndpoint(token) + const client = createRestClient(toHttpUrl(endpoint), workspace, token) + const { model, hierarchy } = await client.getModel() + return new RestClientAdapter(client, hierarchy, model) +} + +/** + * @public + */ +export class Controller { + private readonly workspaces: Map = new Map() + + async get (workspace: WorkspaceUuid): Promise { + let client = this.workspaces.get(workspace) + if (client === undefined) { + client = await WorkspaceClient.create(workspace) + this.workspaces.set(workspace, client) + } + + return client + } + + async close (): Promise { + for (const workspace of this.workspaces.values()) { + await workspace.close() + } + this.workspaces.clear() + } +} + +/** + * @public + */ +export class WorkspaceClient { + private constructor ( + readonly workspace: WorkspaceUuid, + readonly client: Client + ) {} + + static async create (workspace: WorkspaceUuid): Promise { + const token = generateToken(systemAccountUuid, workspace, { service: 'media' }) + const client = await getClient(workspace, token) + return new WorkspaceClient(workspace, client) + } + + async close (): Promise { + await this.client.close() + } + + async updateBlobMetadata (ctx: MeasureContext, result: VideoTranscodeResult, metadata: BlobMetadata): Promise { + if (result.source.source !== BlobSourceType.Doc) { + return + } + + const { objectClass, objectId } = result.source + + const hierarchy = this.client.getHierarchy() + const txOps = new TxOperations(this.client, core.account.System) + + if (hierarchy.isDerived(objectClass, attachment.class.Attachment)) { + const doc = await txOps.findOne(attachment.class.Attachment, { _id: objectId as Ref }) + if (doc !== undefined) { + await txOps.update(doc, { ...doc.metadata, ...metadata }) + } + } + + if (hierarchy.isDerived(objectClass, drive.class.FileVersion)) { + const doc = await txOps.findOne(drive.class.FileVersion, { _id: objectId as Ref }) + if (doc !== undefined) { + await txOps.update(doc, { ...doc.metadata, ...metadata }) + } + } + } +} + +function toHttpUrl (url: string): string { + return url.replace('ws://', 'http://').replace('wss://', 'https://') +} + +class RestClientAdapter implements Client { + constructor ( + private readonly client: RestClient, + private readonly hierarchy: Hierarchy | undefined, + private readonly model: ModelDb | undefined + ) {} + + async domainRequest( + domain: OperationDomain, + params: DomainParams, + options?: DomainRequestOptions + ): Promise> { + throw new Error('Domain request operation not supported') + } + + async findAll( + _class: Ref>, + query: DocumentQuery, + options?: FindOptions + ): Promise> { + return await this.client.findAll(_class, query, options) + } + + async tx (tx: Tx): Promise { + return await this.client.tx(tx) + } + + async findOne( + _class: Ref>, + query: DocumentQuery, + options?: FindOptions + ): Promise | undefined> { + return await this.client.findOne(_class, query, options) + } + + async searchFulltext (query: SearchQuery, options: SearchOptions): Promise { + return await this.client.searchFulltext(query, options) + } + + async close (): Promise { + // No ned to close the REST client + } + + getHierarchy (): Hierarchy { + if (this.hierarchy === undefined) { + throw new Error('Hierarchy is not defined') + } + return this.hierarchy + } + + getModel (): ModelDb { + if (this.model === undefined) { + throw new Error('Model is not defined') + } + return this.model + } +} diff --git a/pods/media/src/config.ts b/pods/media/src/config.ts index d4f90e4974..b733455b85 100644 --- a/pods/media/src/config.ts +++ b/pods/media/src/config.ts @@ -17,13 +17,17 @@ import { config as dotenvConfig } from 'dotenv' dotenvConfig() export interface Config { + AccountsUrl: string Secret: string + ServiceID: string Partitions: number } const config: Config = (() => { const params: Partial = { + AccountsUrl: process.env.ACCOUNTS_URL, Secret: process.env.SECRET, + ServiceID: process.env.SERVICE_ID ?? 'media', Partitions: parseNumber(process.env.PARTITIONS) ?? 1 } diff --git a/pods/media/src/handler.ts b/pods/media/src/handler.ts index 3babbc1abb..2c7be6a88b 100644 --- a/pods/media/src/handler.ts +++ b/pods/media/src/handler.ts @@ -13,16 +13,25 @@ // limitations under the License. // +import attachment, { type Attachment } from '@hcengineering/attachment' +import { Event, MessageEventType } from '@hcengineering/communication-sdk-types' +import drive, { type FileVersion } from '@hcengineering/drive' import core, { type Blob, type Doc, type MeasureContext, - type TxCUD, + type Ref, + type Tx, type TxCreateDoc, - type WorkspaceUuid + type TxDomainEvent, + type WorkspaceUuid, + OperationDomain } from '@hcengineering/core' import { PlatformQueueProducer } from '@hcengineering/server-core' -import { VideoTranscodeRequest, VideoTranscodeResult } from './types' +import { BlobSource, BlobSourceType, VideoTranscodeRequest, VideoTranscodeResult } from './types' +import { WorkspaceClient } from './client' + +const COMMUNICATION = 'communication' as OperationDomain const transcodeIgnoredContentTypes = [ 'video/x-mpegurl', // HLS playlist @@ -36,26 +45,92 @@ function shouldTranscode (contentType: string): boolean { export async function handleTx ( ctx: MeasureContext, workspaceUuid: WorkspaceUuid, - tx: TxCUD, + tx: Tx, producer: PlatformQueueProducer ): Promise { - if (tx.objectClass !== core.class.Blob) return - if (tx._class !== core.class.TxCreateDoc) return + if (tx._class === core.class.TxCreateDoc) { + await handleCreateDocTx(ctx, workspaceUuid, tx as TxCreateDoc, producer) + } else if (tx._class === core.class.TxDomainEvent) { + await handleCommunicationTx(ctx, workspaceUuid, tx as TxDomainEvent, producer) + } +} - const createTx = tx as TxCreateDoc - if (shouldTranscode(createTx.attributes.contentType)) { - const msg: VideoTranscodeRequest = { - workspaceUuid, - blobId: createTx.objectId, - contentType: createTx.attributes.contentType +async function handleCreateDocTx ( + ctx: MeasureContext, + workspaceUuid: WorkspaceUuid, + tx: TxCreateDoc, + producer: PlatformQueueProducer +): Promise { + let blobId: Ref + let contentType: string + + if (tx.objectClass === attachment.class.Attachment || tx.objectClass === attachment.class.Embedding) { + const createTx = tx as TxCreateDoc + blobId = createTx.attributes.file + contentType = createTx.attributes.type + } else if (tx.objectClass === drive.class.FileVersion) { + const createTx = tx as TxCreateDoc + blobId = createTx.attributes.file + contentType = createTx.attributes.type + } else { + return + } + + if (shouldTranscode(contentType)) { + const source: BlobSource = { + source: BlobSourceType.Doc, + objectClass: tx.objectClass, + objectId: tx.objectId } - - ctx.info('Transcode request', { workspaceUuid, msg }) + const msg: VideoTranscodeRequest = { workspaceUuid, blobId, contentType, source } + ctx.info('transcode request', { workspaceUuid, msg }) await producer.send(workspaceUuid, [msg]) } } -export async function handleTranscodeResult (ctx: MeasureContext, msg: VideoTranscodeResult): Promise { - // TODO Handle transcode result - ctx.info('Transcode result', { msg }) +async function handleCommunicationTx ( + ctx: MeasureContext, + workspaceUuid: WorkspaceUuid, + tx: TxDomainEvent, + producer: PlatformQueueProducer +): Promise { + if (tx.domain === COMMUNICATION && tx.event.type === MessageEventType.BlobPatch) { + const event = tx.event + const source: BlobSource = { + source: BlobSourceType.Message, + cardId: event.cardId, + messageId: event.messageId + } + const blobs = event.operations + .filter((it) => it.opcode === 'attach' || it.opcode === 'set') + .flatMap((it) => it.blobs) + const messages: VideoTranscodeRequest[] = blobs.map(({ blobId, mimeType }) => ({ + workspaceUuid, + blobId, + contentType: mimeType, + source + })) + if (messages.length > 0) { + await producer.send(workspaceUuid, messages) + } + } +} + +export async function handleTranscodeResult ( + ctx: MeasureContext, + workspaceUuid: WorkspaceUuid, + msg: VideoTranscodeResult +): Promise { + ctx.info('transcode result', { workspaceUuid, msg }) + const metadata = { + hls: { + source: msg.playlist, + thumbnail: msg.thumbnail + } + } + + if (msg.source !== undefined && msg.source.source === BlobSourceType.Doc) { + const client = await WorkspaceClient.create(workspaceUuid) + await client.updateBlobMetadata(ctx, msg, metadata) + } } diff --git a/pods/media/src/index.ts b/pods/media/src/index.ts index 4683095ce7..a95e309524 100644 --- a/pods/media/src/index.ts +++ b/pods/media/src/index.ts @@ -18,6 +18,7 @@ import { configureAnalytics, SplitLogger } from '@hcengineering/analytics-servic import { Doc, MeasureMetricsContext, TxCUD, newMetrics } from '@hcengineering/core' import { getPlatformQueue } from '@hcengineering/kafka' import { setMetadata } from '@hcengineering/platform' +import serverClient from '@hcengineering/server-client' import { initStatisticsContext, QueueTopic } from '@hcengineering/server-core' import serverToken from '@hcengineering/server-token' import { join } from 'path' @@ -33,6 +34,8 @@ const topicTranscodeResult = 'stream.transcode.result' const setupMetadata = (): void => { setMetadata(serverToken.metadata.Secret, config.Secret) setMetadata(serverToken.metadata.Service, 'media') + setMetadata(serverClient.metadata.Endpoint, config.AccountsUrl) + setMetadata(serverClient.metadata.UserAgent, config.ServiceID) } async function main (): Promise { @@ -63,7 +66,7 @@ async function main (): Promise { queue.createConsumer(ctx, topicTranscodeResult, application, async (msgs) => { for (const msg of msgs) { for (const res of msg.value) { - await handleTranscodeResult(ctx, res) + await handleTranscodeResult(ctx, msg.workspace, res) } } }) diff --git a/pods/media/src/types.ts b/pods/media/src/types.ts index 09c9ee4965..b410809bf5 100644 --- a/pods/media/src/types.ts +++ b/pods/media/src/types.ts @@ -13,15 +13,39 @@ // limitations under the License. // +import type { Class, Doc, Ref } from '@hcengineering/core' +import type { CardID, MessageID } from '@hcengineering/communication-types' + +export enum BlobSourceType { + Doc = 'doc', + Message = 'message' +} + +export interface BlobSourceDoc { + source: BlobSourceType.Doc + objectClass: Ref> + objectId: string +} + +export interface BlobSourceMessage { + source: BlobSourceType.Message + cardId: CardID + messageId: MessageID +} + +export type BlobSource = BlobSourceDoc | BlobSourceMessage + export interface VideoTranscodeRequest { blobId: string workspaceUuid: string contentType: string + source: BlobSource } export interface VideoTranscodeResult { blobId: string workspaceUuid: string + source: BlobSource thumbnail?: string playlist?: string