diff --git a/.env.example b/.env.example index 7c876d1..14b6bdc 100644 --- a/.env.example +++ b/.env.example @@ -245,3 +245,6 @@ EVOO_COCKPIT_CORS_ORIGIN= # MEDIA_WORKER_TOKEN=REPLACE_WITH_RANDOM_TOKEN # Private platform profiles that override the shipped ones. Keep outside the repo. # MEDIA_PROFILES_DIR=~/.evoo/media-profiles +# Media jobs per worker process: model inference (GPU) and ffmpeg work (CPU). +# MEDIA_GPU_CONCURRENCY=1 +# MEDIA_CPU_CONCURRENCY=2 diff --git a/apps/api/src/app.ts b/apps/api/src/app.ts index 948d735..047c6a7 100644 --- a/apps/api/src/app.ts +++ b/apps/api/src/app.ts @@ -27,6 +27,8 @@ import { createJobReconciler, createMessagingContextLoader, createMessagingTeamChatSender, + createMediaJobHandlers, + createMediaServices, createProfileConsolidationScheduler, createRunExecutor, createRunSandbox, @@ -110,6 +112,7 @@ import { TEAM_CHAT_STARTUP_SHUTDOWN_MS, } from "./team-chat-startup.js"; import { mountTradingBotStatus } from "./trading-bot-proxy.js"; +import { mountMediaHttpRoutes } from "./media-http.js"; import { mountVoiceHttpRoutes } from "./voice.js"; import { mountWebhookHttpRoutes } from "./webhook.js"; @@ -414,6 +417,18 @@ export async function createApp( shutdownSignal: shutdown.signal, }); + const mediaServices = createMediaServices({ + config: env.media, + prisma, + jobs, + artifacts, + runtime, + realtime, + resolveModel: (scope) => executor.resolveModel(scope), + }); + const mediaJobs = mediaServices + ? createMediaJobHandlers(mediaServices, { signal: shutdown.signal }) + : null; const jobHandlers = createBackgroundJobHandlers({ executor, prisma, @@ -421,6 +436,7 @@ export async function createApp( home, jobs, events, + ...(mediaJobs ? { media: mediaJobs.handlers } : {}), workerId: "api", runtime, secretStore: secrets, @@ -438,6 +454,9 @@ export async function createApp( }); await graphileWorkerHost.start(jobHandlers); } + if (mediaJobs) { + void mediaJobs.resume().catch((error) => getLogger().error("media ingest resume failed", error)); + } const reconciler = inMemoryJobs ? createJobReconciler({ prisma, @@ -469,6 +488,7 @@ export async function createApp( remoteConnectors, artifacts, dataDir: env.dataDir, + mediaServices, messaging: { enabled: Boolean(messaging), providers: messaging?.platforms().map((platform) => platform.provider) ?? [], @@ -583,6 +603,13 @@ export async function createApp( if (actor) enrichLogContext({ "user.id": actor.userId, "space.id": actor.spaceId }); return actor; }); + mountMediaHttpRoutes(app, { services: mediaServices }, async (c) => { + const session = await auth.api.getSession({ headers: sessionHeaders(c.req.raw) }); + if (!session?.user) return null; + return requireMembership(prisma, session.user.id, c.req.header("x-rakazo-space-id")).catch( + () => null, + ); + }); mountWebhookHttpRoutes(app, { prisma, secrets, events, jobs }); // Shared with stop so a shutdown during retry delays does not restart polling. let messagingStopped = false; diff --git a/apps/api/src/env.ts b/apps/api/src/env.ts index c88c2da..a520691 100644 --- a/apps/api/src/env.ts +++ b/apps/api/src/env.ts @@ -1,4 +1,6 @@ +import type { MediaConfig } from "@rakazo/adapters"; import { + mediaConfigFromEnv, resolveCloudAgentProvider, resolveDeploymentModel, resolveSandboxProvider, @@ -15,6 +17,8 @@ export { resolveCloudAgentProvider, resolveSandboxProvider } from "@rakazo/adapt export interface AppEnv { nodeEnv: string; + /** Media Studio; disabled unless a sidecar (or the scripted emulator) is configured. */ + media: MediaConfig; desktopStackToken?: string; databaseUrl: string; realtimeDatabaseUrl: string; @@ -106,6 +110,7 @@ export function loadEnv(source: NodeJS.ProcessEnv = process.env): AppEnv { const updaterToken = optional(source.RAKAZO_UPDATER_TOKEN); return { nodeEnv: source.NODE_ENV ?? "", + media: mediaConfigFromEnv(source), databaseUrl: required(source, "DATABASE_URL"), realtimeDatabaseUrl: source.REALTIME_DATABASE_URL ?? required(source, "DATABASE_URL"), desktopStackToken: optional(source.RAKAZO_DESKTOP_STACK_TOKEN), diff --git a/apps/api/src/router.ts b/apps/api/src/router.ts index b69c2a9..b032817 100644 --- a/apps/api/src/router.ts +++ b/apps/api/src/router.ts @@ -24,6 +24,7 @@ import type { ConnectorRegistry, EncryptedSecretStore, IntegrationProviderSettings, + MediaServices, MemoryProviderResolver, PiOAuthLogins, RemoteConnectorDependencies, @@ -464,6 +465,8 @@ export interface RouterDeps { remoteConnectors?: RemoteConnectorDependencies; artifacts: ArtifactStore; dataDir: string; + /** Null when no media sidecar (or scripted emulator) is configured; Media Studio then stays hidden. */ + mediaServices?: MediaServices | null; /** Present when the external messaging surface is enabled. */ messaging?: { enabled: boolean; providers: string[]; openSignup: boolean }; env: { @@ -527,7 +530,11 @@ export function createRouter(deps: RouterDeps) { }); return os.router({ - media: createMediaRouter({ prisma: deps.prisma, dataDir: deps.dataDir }), + media: createMediaRouter({ + prisma: deps.prisma, + services: deps.mediaServices ?? null, + dataDir: deps.dataDir, + }), aiConsent: { status: authed.aiConsent.status.handler(({ context, input }) => aiConsentStatus(deps, context.actor, input), diff --git a/apps/worker/src/index.ts b/apps/worker/src/index.ts index 804281c..9524a58 100644 --- a/apps/worker/src/index.ts +++ b/apps/worker/src/index.ts @@ -12,6 +12,8 @@ import { createJobReconciler, createMessagingContextLoader, createPostgresReconciliationLeadership, + createMediaJobHandlers, + createMediaServices, createProfileConsolidationScheduler, createRunExecutor, createRunSandbox, @@ -29,6 +31,7 @@ import { isPipedreamEnabled, LocalAgentHomeStore, LocalArtifactStore, + mediaConfigFromEnv, McpConnector, McpOAuthBroker, messagingEnvFromProcess, @@ -201,6 +204,25 @@ async function main() { cloudAgent, }); + const mediaAbort = new AbortController(); + const mediaServices = createMediaServices({ + config: mediaConfigFromEnv(process.env), + prisma, + jobs, + artifacts, + runtime, + realtime, + resolveModel: (scope) => executor.resolveModel(scope), + }); + const mediaJobs = mediaServices + ? createMediaJobHandlers(mediaServices, { + signal: mediaAbort.signal, + limits: { + gpu: parsePositiveInteger(process.env.MEDIA_GPU_CONCURRENCY, 1), + cpu: parsePositiveInteger(process.env.MEDIA_CPU_CONCURRENCY, 2), + }, + }) + : null; const jobHandlers = createBackgroundJobHandlers({ executor, prisma, @@ -208,6 +230,7 @@ async function main() { home, jobs, events, + ...(mediaJobs ? { media: mediaJobs.handlers } : {}), workerId: process.pid.toString(), runtime, secretStore: secrets, @@ -235,6 +258,9 @@ async function main() { await new Promise((resolve) => setTimeout(resolve, databaseCapacityBackoffMs(attempt))); } } + if (mediaJobs) { + void mediaJobs.resume().catch((error) => logger.error("media ingest resume failed", error)); + } const reconciler = createJobReconciler({ prisma, jobs, @@ -253,6 +279,7 @@ async function main() { stopping = true; try { await reconciler.stop(); + mediaAbort.abort(); await jobHost.stop(); await jobs.close(); await realtime.close(); diff --git a/packages/adapters/src/index.ts b/packages/adapters/src/index.ts index fbc87eb..bf5df64 100644 --- a/packages/adapters/src/index.ts +++ b/packages/adapters/src/index.ts @@ -85,6 +85,8 @@ export * from "./media-factory.js"; export * from "./media-ffmpeg-renderer.js"; export * from "./media-folder-source.js"; export * from "./media-folder-source-probe.js"; +export * from "./media-ingest.js"; +export { defaultMediaJobLimits, MediaJobLimiter, type MediaJobLimits } from "./media-jobs.js"; export * from "./media-marketing.js"; export * from "./media-marketing-copy.js"; export * from "./media-marketing-files.js"; @@ -103,6 +105,7 @@ export { } from "./media-motion-renderer.js"; export * from "./media-publish.js"; export * from "./media-publisher-bundle.js"; +export * from "./media-services.js"; export { createMediaRenderHandler, createMediaRenderService, diff --git a/packages/adapters/src/media-services.ts b/packages/adapters/src/media-services.ts new file mode 100644 index 0000000..9eaea81 --- /dev/null +++ b/packages/adapters/src/media-services.ts @@ -0,0 +1,136 @@ +import type { + AgentRunModel, + AgentRuntime, + ArtifactStore, + JobPublisher, + MediaJobHandlers, + RealtimeFanout, +} from "@rakazo/adapter-kit"; +import type { ModelTier } from "@rakazo/contracts"; +import type { PrismaClient } from "@rakazo/db"; +import { createMediaRepos } from "@rakazo/db"; +import { getLogger } from "@rakazo/logging"; +import type { ProfileSet } from "@rakazo/media"; +import { mergeProfiles } from "@rakazo/media"; +import { BUILTIN_PROFILES_DIR, loadProfilesFromDir } from "@rakazo/media/node"; +import type { MediaServiceDeps, MediaServices } from "./media-context.js"; +import { createMediaModelClient, mediaTopic } from "./media-context.js"; +import { + createMediaInferenceProvider, + createMediaSourceProvider, + type MediaConfig, + mediaEmulatorEnabled, +} from "./media-factory.js"; +import { FfmpegMediaRenderer } from "./media-ffmpeg-renderer.js"; +import { + createMediaIngestHandlers, + createMediaIngestService, + resumeMediaIngest, +} from "./media-ingest.js"; +import type { MediaJobLimits } from "./media-jobs.js"; +import { createMediaMarketingJobHandlers, createMediaMarketingService } from "./media-marketing.js"; +import { ResvgMotionGraphicsRenderer } from "./media-motion-renderer.js"; +import { createMediaPrivacyService } from "./media-privacy.js"; +import type { MediaUploadReminder } from "./media-publish.js"; +import { BundleMediaPublisher } from "./media-publisher-bundle.js"; +import { createMediaRenderHandler, createMediaRenderService } from "./media-render.js"; + +const PROFILE_CACHE_MS = 60_000; + +/** + * Composition root for Media Studio, shared by the API and the worker. Returns + * null when no sidecar (or scripted emulator) is configured; media then stays + * hidden in every client. + */ +export function createMediaServices(input: { + config: MediaConfig; + prisma: PrismaClient; + jobs: JobPublisher; + artifacts: ArtifactStore; + runtime: AgentRuntime; + realtime: RealtimeFanout | null; + resolveModel: (scope: { + userId: string; + spaceId: string; + tier?: ModelTier; + }) => Promise; + env?: NodeJS.ProcessEnv; +}): MediaServices | null { + const env = input.env ?? process.env; + if (!input.config.enabled) return null; + const inference = createMediaInferenceProvider(env); + if (!inference) return null; + const sources: MediaServiceDeps["sources"] = { folder: createMediaSourceProvider("folder", env) }; + if (mediaEmulatorEnabled(env)) sources.emulator = createMediaSourceProvider("emulator", env); + + let cached: { at: number; set: ProfileSet } | null = null; + const profiles = async (): Promise => { + if (cached && Date.now() - cached.at < PROFILE_CACHE_MS) return cached.set; + const [builtin, own] = await Promise.all([ + loadProfilesFromDir(BUILTIN_PROFILES_DIR), + loadProfilesFromDir(input.config.profilesDir), + ]); + const set = mergeProfiles(builtin.profiles, own.profiles); + set.errors = [...builtin.errors, ...own.errors]; + for (const error of set.errors) getLogger().warn("media profile ignored", error); + cached = { at: Date.now(), set }; + return set; + }; + + const deps: MediaServiceDeps = { + prisma: input.prisma, + repos: createMediaRepos(input.prisma), + jobs: input.jobs, + inference, + renderer: new FfmpegMediaRenderer({ root: input.config.mediaRoot }), + motion: new ResvgMotionGraphicsRenderer(), + publisher: new BundleMediaPublisher(), + shortener: null, + sources, + mediaRoot: input.config.mediaRoot, + profiles, + artifacts: input.artifacts, + model: createMediaModelClient({ + prisma: input.prisma, + runtime: input.runtime, + resolveModel: input.resolveModel, + }), + publish: async (spaceId, event) => { + await input.realtime?.publish(mediaTopic(spaceId), JSON.stringify(event)); + }, + now: () => new Date(), + }; + const privacy = createMediaPrivacyService(deps); + return { + deps, + ingest: createMediaIngestService(deps), + privacy, + render: createMediaRenderService(deps), + marketing: createMediaMarketingService(deps), + }; +} + +/** + * Every media job handler for one job host. Any process that runs a job host + * (worker, or the API in single-process mode) must register these, otherwise + * it would acknowledge media jobs without running them. + */ +export function createMediaJobHandlers( + services: MediaServices, + options: { signal?: AbortSignal; limits?: MediaJobLimits; remind?: MediaUploadReminder } = {}, +): { handlers: MediaJobHandlers; resume: () => Promise } { + const ingestOptions = { + ...(options.signal ? { signal: options.signal } : {}), + ...(options.limits ? { limits: options.limits } : {}), + }; + return { + handlers: { + ...createMediaIngestHandlers(services.deps, ingestOptions), + "media.render": createMediaRenderHandler(services.deps, services.privacy), + ...createMediaMarketingJobHandlers(services.deps, services.marketing, { + ...(options.remind ? { remind: options.remind } : {}), + }), + }, + resume: () => resumeMediaIngest(services.deps, options.limits), + }; +} diff --git a/packages/testkit/src/cli/harness.ts b/packages/testkit/src/cli/harness.ts index 27415a4..311e2fa 100644 --- a/packages/testkit/src/cli/harness.ts +++ b/packages/testkit/src/cli/harness.ts @@ -117,6 +117,12 @@ async function main() { "packages/adapters/src/cloud-agent.postgres.test.ts", "packages/adapters/src/builtin-memory-provider.postgres.test.ts", "packages/adapters/src/profile-consolidation.postgres.test.ts", + "packages/db/src/media.postgres.test.ts", + "packages/db/src/media-isolation.postgres.test.ts", + "packages/db/src/media-production.postgres.test.ts", + "packages/adapters/src/media-ingest.postgres.test.ts", + "packages/adapters/src/media-marketing.postgres.test.ts", + "apps/api/src/media.postgres.test.ts", ]; // Each app reconciles all durable work in its database, including intentionally // unfinished fixture runs. Clone the pristine migrated schema so one suite