UBERF-12153 Transcode only used blobs (#9456)

Signed-off-by: Alexander Onnikov <Alexander.Onnikov@xored.com>

.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"],

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:

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"
   }

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<Client> {
+  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<WorkspaceUuid, WorkspaceClient> = new Map<WorkspaceUuid, WorkspaceClient>()
+
+  async get (workspace: WorkspaceUuid): Promise<WorkspaceClient> {
+    let client = this.workspaces.get(workspace)
+    if (client === undefined) {
+      client = await WorkspaceClient.create(workspace)
+      this.workspaces.set(workspace, client)
+    }
+
+    return client
+  }
+
+  async close (): Promise<void> {
+    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<WorkspaceClient> {
+    const token = generateToken(systemAccountUuid, workspace, { service: 'media' })
+    const client = await getClient(workspace, token)
+    return new WorkspaceClient(workspace, client)
+  }
+
+  async close (): Promise<void> {
+    await this.client.close()
+  }
+
+  async updateBlobMetadata (ctx: MeasureContext, result: VideoTranscodeResult, metadata: BlobMetadata): Promise<void> {
+    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<Attachment> })
+      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<FileVersion> })
+      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<T>(
+    domain: OperationDomain,
+    params: DomainParams,
+    options?: DomainRequestOptions
+  ): Promise<DomainResult<T>> {
+    throw new Error('Domain request operation not supported')
+  }
+
+  async findAll<T extends Doc>(
+    _class: Ref<Class<T>>,
+    query: DocumentQuery<T>,
+    options?: FindOptions<T>
+  ): Promise<FindResult<T>> {
+    return await this.client.findAll(_class, query, options)
+  }
+
+  async tx (tx: Tx): Promise<TxResult> {
+    return await this.client.tx(tx)
+  }
+
+  async findOne<T extends Doc>(
+    _class: Ref<Class<T>>,
+    query: DocumentQuery<T>,
+    options?: FindOptions<T>
+  ): Promise<WithLookup<T> | undefined> {
+    return await this.client.findOne(_class, query, options)
+  }
+
+  async searchFulltext (query: SearchQuery, options: SearchOptions): Promise<SearchResult> {
+    return await this.client.searchFulltext(query, options)
+  }
+
+  async close (): Promise<void> {
+    // 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
+  }
+}

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<Config> = {
+    AccountsUrl: process.env.ACCOUNTS_URL,
     Secret: process.env.SECRET,
+    ServiceID: process.env.SERVICE_ID ?? 'media',
     Partitions: parseNumber(process.env.PARTITIONS) ?? 1
   }
 

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<Doc>,
+  tx: Tx,
   producer: PlatformQueueProducer<VideoTranscodeRequest>
 ): Promise<void> {
-  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<Doc>, producer)
+  } else if (tx._class === core.class.TxDomainEvent) {
+    await handleCommunicationTx(ctx, workspaceUuid, tx as TxDomainEvent<Event>, producer)
+  }
+}
 
-  const createTx = tx as TxCreateDoc<Blob>
-  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<Doc>,
+  producer: PlatformQueueProducer<VideoTranscodeRequest>
+): Promise<void> {
+  let blobId: Ref<Blob>
+  let contentType: string
+
+  if (tx.objectClass === attachment.class.Attachment || tx.objectClass === attachment.class.Embedding) {
+    const createTx = tx as TxCreateDoc<Attachment>
+    blobId = createTx.attributes.file
+    contentType = createTx.attributes.type
+  } else if (tx.objectClass === drive.class.FileVersion) {
+    const createTx = tx as TxCreateDoc<FileVersion>
+    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<void> {
-  // TODO Handle transcode result
-  ctx.info('Transcode result', { msg })
+async function handleCommunicationTx (
+  ctx: MeasureContext,
+  workspaceUuid: WorkspaceUuid,
+  tx: TxDomainEvent<Event>,
+  producer: PlatformQueueProducer<VideoTranscodeRequest>
+): Promise<void> {
+  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<void> {
+  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)
+  }
 }

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<void> {
@@ -63,7 +66,7 @@ async function main (): Promise<void> {
   queue.createConsumer<VideoTranscodeResult>(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)
       }
     }
   })

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<Class<Doc>>
+  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