fix(server): realtime module not loaded (#14952)

#### PR Dependency Tree


* **PR #14952** 👈

This tree was auto-generated by
[Charcoal](https://github.com/danerwilliams/charcoal)

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->

## Summary by CodeRabbit

* **Refactor**
* Optimized workspace invite link fetching by separating it from general
workspace configuration queries for improved performance.
* Reorganized transcription-related backend modules to better separate
concerns and enable real-time functionality.

* **Chores**
* Updated generated GraphQL types and iOS query definitions to reflect
API changes.

[![Review Change
Stack](https://storage.googleapis.com/coderabbit_public_assets/review-stack-in-coderabbit-ui.svg)](https://app.coderabbit.ai/change-stack/toeverything/AFFiNE/pull/14952)

<!-- end of auto-generated comment: release notes by coderabbit.ai -->
This commit is contained in:
DarkSky
2026-05-12 18:54:42 +08:00
committed by GitHub
parent ac6d0d35af
commit a1d150a748
18 changed files with 246 additions and 104 deletions
+3 -2
View File
@@ -55,7 +55,7 @@ import { Env } from './env';
import { ModelsModule } from './models';
import { CalendarModule } from './plugins/calendar';
import { CaptchaModule } from './plugins/captcha';
import { CopilotModule } from './plugins/copilot';
import { CopilotModule, CopilotRealtimeModule } from './plugins/copilot';
import { CustomerIoModule } from './plugins/customerio';
import { GCloudModule } from './plugins/gcloud';
import { IndexerModule } from './plugins/indexer';
@@ -185,7 +185,8 @@ export function buildAppModule(env: Env) {
.useIf(
() => env.flavors.sync || env.flavors.front,
SyncModule,
TelemetryModule
TelemetryModule,
CopilotRealtimeModule
)
// graphql server only
.useIf(
@@ -2,7 +2,10 @@ import { getRealtimeInputKey } from '@affine/realtime';
import test from 'ava';
import { z } from 'zod';
import type { CopilotTranscriptionReader } from '../../../plugins/copilot/transcript';
import { CopilotTranscriptRealtimeProvider } from '../../../plugins/copilot/transcript';
import type { CurrentUser } from '../../auth';
import type { AccessController } from '../../permission';
import { RealtimeGateway } from '../gateway';
import {
realtimeCommentRoom,
@@ -191,6 +194,61 @@ test('registerRealtimeLiveQuery registers paired request and topic handlers', as
);
});
test('copilot transcript realtime provider registers task live query handlers', async t => {
const registry = new RealtimeRegistry();
const assertions: unknown[] = [];
const ac = {
user(userId: string) {
return {
workspace(workspaceId: string) {
return {
allowLocal() {
return this;
},
async assert(action: string) {
assertions.push({ userId, workspaceId, action });
},
};
},
};
},
} as unknown as AccessController;
const transcript = {
async queryTask(
userId: string,
workspaceId: string,
taskId?: string,
blobId?: string
) {
return { id: taskId ?? blobId, status: 'finished', userId, workspaceId };
},
} as unknown as CopilotTranscriptionReader;
new CopilotTranscriptRealtimeProvider(
ac,
transcript,
registry
).onModuleInit();
t.deepEqual(
await registry.getRequest('copilot.transcript.task.get').handle(user, {
workspaceId: 'space',
taskId: 'task',
}),
{
task: {
id: 'task',
status: 'finished',
userId: 'u1',
workspaceId: 'space',
},
}
);
t.deepEqual(assertions, [
{ userId: 'u1', workspaceId: 'space', action: 'Workspace.Copilot' },
]);
});
test('publisher emits realtime event with shared input key', t => {
const registry = new RealtimeRegistry();
registry.registerTopic({
@@ -16,6 +16,7 @@ import {
COPILOT_API_PROVIDERS,
COPILOT_FEATURE_PROVIDERS,
COPILOT_KERNEL_PROVIDERS,
COPILOT_TRANSCRIPT_REALTIME_PROVIDERS,
} from './module-providers';
const COPILOT_SHARED_IMPORTS = [
@@ -36,6 +37,12 @@ const COPILOT_SHARED_IMPORTS = [
})
export class CopilotKernelModule {}
@Module({
imports: [PermissionModule],
providers: [...COPILOT_TRANSCRIPT_REALTIME_PROVIDERS],
})
export class CopilotRealtimeModule {}
@Module({
imports: [...COPILOT_SHARED_IMPORTS, CopilotKernelModule],
providers: [...COPILOT_FEATURE_PROVIDERS],
@@ -54,6 +54,7 @@ import { TurnOrchestrator } from './runtime/turn-orchestrator';
import { ChatSessionService } from './session';
import { CopilotStorage } from './storage';
import {
CopilotTranscriptionReader,
CopilotTranscriptionResolver,
CopilotTranscriptionService,
CopilotTranscriptRealtimeProvider,
@@ -113,10 +114,15 @@ export const COPILOT_CONTEXT_PROVIDERS = [
CopilotEmbeddingRealtimeProvider,
];
export const COPILOT_TRANSCRIPT_REALTIME_PROVIDERS = [
CopilotTranscriptionReader,
CopilotTranscriptRealtimeProvider,
];
export const COPILOT_TRANSCRIPT_PROVIDERS = [
CopilotTranscriptionService,
CopilotTranscriptionResolver,
CopilotTranscriptRealtimeProvider,
...COPILOT_TRANSCRIPT_REALTIME_PROVIDERS,
];
export const COPILOT_WORKSPACE_PROVIDERS = [
@@ -1,3 +1,5 @@
export type { TranscriptionJob } from './job';
export { CopilotTranscriptionReader } from './reader';
export { CopilotTranscriptRealtimeProvider } from './realtime';
export { CopilotTranscriptionResolver } from './resolver';
export { CopilotTranscriptionService } from './service';
@@ -0,0 +1,52 @@
import { AiJobStatus } from '@prisma/client';
import { TranscriptPayloadSchema } from './schema';
import type { AudioBlobInfos, TranscriptionPayload } from './types';
export type TranscriptionJob = {
id: string;
status: AiJobStatus;
infos?: AudioBlobInfos;
transcription?: TranscriptionPayload;
};
export function taskStatusToPublicStatus(status: string): AiJobStatus {
switch (status) {
case 'pending':
return AiJobStatus.pending;
case 'running':
return AiJobStatus.running;
case 'ready':
case 'settled':
return AiJobStatus.finished;
default:
return AiJobStatus.failed;
}
}
export function taskToJob(
task: {
id: string;
status: string;
protectedResult: unknown;
} | null,
mapStatus: (status: string) => AiJobStatus = taskStatusToPublicStatus
): TranscriptionJob | null {
if (!task) {
return null;
}
const status = mapStatus(task.status);
const ret: TranscriptionJob = {
id: task.id,
status,
};
if (task.protectedResult) {
const parsed = TranscriptPayloadSchema.safeParse(task.protectedResult);
ret.infos = parsed.success ? (parsed.data.infos ?? []) : [];
if (task.status === 'settled' && parsed.success) {
ret.transcription = parsed.data;
}
}
return ret;
}
@@ -0,0 +1,24 @@
import { Injectable } from '@nestjs/common';
import { Models } from '../../../models';
import { taskToJob } from './job';
@Injectable()
export class CopilotTranscriptionReader {
constructor(private readonly models: Models) {}
async queryTask(
userId: string,
workspaceId: string,
taskId?: string,
blobId?: string
) {
const task = await this.models.copilotTranscriptTask.getWithUser(
userId,
workspaceId,
taskId,
blobId
);
return taskToJob(task);
}
}
@@ -8,13 +8,13 @@ import {
realtimeTranscriptTaskRoom,
registerRealtimeLiveQuery,
} from '../../../core/realtime';
import { CopilotTranscriptionService } from './service';
import { CopilotTranscriptionReader } from './reader';
@Injectable()
export class CopilotTranscriptRealtimeProvider implements OnModuleInit {
constructor(
private readonly ac: AccessController,
private readonly transcript: CopilotTranscriptionService,
private readonly transcript: CopilotTranscriptionReader,
@Optional() private readonly registry?: RealtimeRegistry
) {}
@@ -23,8 +23,9 @@ import {
import { CurrentUser } from '../../../core/auth';
import { AccessController } from '../../../core/permission';
import { CopilotType } from '../resolver';
import type { TranscriptionJob } from './job';
import { buildLegacyProjection } from './projection';
import { CopilotTranscriptionService, TranscriptionJob } from './service';
import { CopilotTranscriptionService } from './service';
import type {
AudioSliceManifestItem,
MeetingActionItem,
@@ -20,13 +20,13 @@ import { CopilotProviderType } from '../providers/types';
import { ActionRuntimeBridge } from '../runtime/action-runtime-bridge';
import { TaskPolicy } from '../runtime/task-policy';
import { CopilotStorage } from '../storage';
import { taskToJob, type TranscriptionJob } from './job';
import {
TranscriptActionResultContract,
TranscriptPayloadSchema,
} from './schema';
import type {
AudioBlobInfos,
TranscriptionPayload,
TranscriptionPayloadV2,
TranscriptionSubmitInput,
} from './types';
@@ -36,27 +36,6 @@ const TRANSCRIPT_ACTION_ID = 'transcript.audio.gemini';
const TRANSCRIPT_ACTION_VERSION = 'v1';
const TRANSCRIPT_STRATEGY = 'gemini';
export type TranscriptionJob = {
id: string;
status: AiJobStatus;
infos?: AudioBlobInfos;
transcription?: TranscriptionPayload;
};
function taskStatusToPublicStatus(status: string): AiJobStatus {
switch (status) {
case 'pending':
return AiJobStatus.pending;
case 'running':
return AiJobStatus.running;
case 'ready':
case 'settled':
return AiJobStatus.finished;
default:
return AiJobStatus.failed;
}
}
@Injectable()
export class CopilotTranscriptionService {
constructor(
@@ -85,33 +64,6 @@ export class CopilotTranscriptionService {
};
}
private taskToJob(
task: {
id: string;
status: string;
protectedResult: unknown;
} | null,
mapStatus: (status: string) => AiJobStatus = taskStatusToPublicStatus
): TranscriptionJob | null {
if (!task) {
return null;
}
const status = mapStatus(task.status);
const ret: TranscriptionJob = {
id: task.id,
status,
};
if (task.protectedResult) {
const parsed = TranscriptPayloadSchema.safeParse(task.protectedResult);
ret.infos = parsed.success ? (parsed.data.infos ?? []) : [];
if (task.status === 'settled' && parsed.success) {
ret.transcription = parsed.data;
}
}
return ret;
}
private async resolveTranscriptStrategy(userId: string, strategy?: string) {
if (strategy && strategy !== TRANSCRIPT_STRATEGY) {
throw new BadRequestException(
@@ -327,7 +279,7 @@ export class CopilotTranscriptionService {
}
if (task.status === 'settled') {
return this.taskToJob(task);
return taskToJob(task);
}
await this.access?.assertQuotaOrByok({
@@ -337,7 +289,7 @@ export class CopilotTranscriptionService {
});
const settled = await this.models.copilotTranscriptTask.settle(task.id);
return this.taskToJob(settled);
return taskToJob(settled);
}
async queryTask(
@@ -352,10 +304,7 @@ export class CopilotTranscriptionService {
taskId,
blobId
);
if (task) {
return this.taskToJob(task);
}
return null;
return taskToJob(task);
}
@OnJob('copilot.transcript.task.submit')
+13 -4
View File
@@ -3126,10 +3126,6 @@ export const getWorkspaceConfigQuery = {
enableSharing
enableUrlPreview
enableDocEmbedding
inviteLink {
link
expireTime
}
}
}`,
};
@@ -3195,6 +3191,19 @@ export const acceptInviteByInviteIdMutation = {
}`,
};
export const getWorkspaceInviteLinkQuery = {
id: 'getWorkspaceInviteLinkQuery' as const,
op: 'getWorkspaceInviteLink',
query: `query getWorkspaceInviteLink($id: String!) {
workspace(id: $id) {
inviteLink {
link
expireTime
}
}
}`,
};
export const createInviteLinkMutation = {
id: 'createInviteLinkMutation' as const,
op: 'createInviteLink',
@@ -4,9 +4,5 @@ query getWorkspaceConfig($id: String!) {
enableSharing
enableUrlPreview
enableDocEmbedding
inviteLink {
link
expireTime
}
}
}
@@ -0,0 +1,8 @@
query getWorkspaceInviteLink($id: String!) {
workspace(id: $id) {
inviteLink {
link
expireTime
}
}
}
+21 -5
View File
@@ -7794,11 +7794,6 @@ export type GetWorkspaceConfigQuery = {
enableSharing: boolean;
enableUrlPreview: boolean;
enableDocEmbedding: boolean;
inviteLink: {
__typename?: 'InviteLink';
link: string;
expireTime: string;
} | null;
};
};
@@ -7867,6 +7862,22 @@ export type AcceptInviteByInviteIdMutation = {
acceptInviteById: boolean;
};
export type GetWorkspaceInviteLinkQueryVariables = Exact<{
id: Scalars['String']['input'];
}>;
export type GetWorkspaceInviteLinkQuery = {
__typename?: 'Query';
workspace: {
__typename?: 'WorkspaceType';
inviteLink: {
__typename?: 'InviteLink';
link: string;
expireTime: string;
} | null;
};
};
export type CreateInviteLinkMutationVariables = Exact<{
workspaceId: Scalars['String']['input'];
expireTime: WorkspaceInviteLinkExpireTime;
@@ -8418,6 +8429,11 @@ export type Queries =
variables: GetWorkspaceConfigQueryVariables;
response: GetWorkspaceConfigQuery;
}
| {
name: 'getWorkspaceInviteLinkQuery';
variables: GetWorkspaceInviteLinkQueryVariables;
response: GetWorkspaceInviteLinkQuery;
}
| {
name: 'workspaceInvoicesQuery';
variables: WorkspaceInvoicesQueryVariables;
@@ -7,7 +7,7 @@ public class GetWorkspaceConfigQuery: GraphQLQuery {
public static let operationName: String = "getWorkspaceConfig"
public static let operationDocument: ApolloAPI.OperationDocument = .init(
definition: .init(
#"query getWorkspaceConfig($id: String!) { workspace(id: $id) { __typename enableAi enableSharing enableUrlPreview enableDocEmbedding inviteLink { __typename link expireTime } } }"#
#"query getWorkspaceConfig($id: String!) { workspace(id: $id) { __typename enableAi enableSharing enableUrlPreview enableDocEmbedding } }"#
))
public var id: String
@@ -47,7 +47,6 @@ public class GetWorkspaceConfigQuery: GraphQLQuery {
.field("enableSharing", Bool.self),
.field("enableUrlPreview", Bool.self),
.field("enableDocEmbedding", Bool.self),
.field("inviteLink", InviteLink?.self),
] }
public static var __fulfilledFragments: [any ApolloAPI.SelectionSet.Type] { [
GetWorkspaceConfigQuery.Data.Workspace.self
@@ -61,31 +60,6 @@ public class GetWorkspaceConfigQuery: GraphQLQuery {
public var enableUrlPreview: Bool { __data["enableUrlPreview"] }
/// Enable doc embedding
public var enableDocEmbedding: Bool { __data["enableDocEmbedding"] }
/// invite link for workspace
public var inviteLink: InviteLink? { __data["inviteLink"] }
/// Workspace.InviteLink
///
/// Parent Type: `InviteLink`
public struct InviteLink: AffineGraphQL.SelectionSet {
public let __data: DataDict
public init(_dataDict: DataDict) { __data = _dataDict }
public static var __parentType: any ApolloAPI.ParentType { AffineGraphQL.Objects.InviteLink }
public static var __selections: [ApolloAPI.Selection] { [
.field("__typename", String.self),
.field("link", String.self),
.field("expireTime", AffineGraphQL.DateTime.self),
] }
public static var __fulfilledFragments: [any ApolloAPI.SelectionSet.Type] { [
GetWorkspaceConfigQuery.Data.Workspace.InviteLink.self
] }
/// Invite link
public var link: String { __data["link"] }
/// Invite link expire time
public var expireTime: AffineGraphQL.DateTime { __data["expireTime"] }
}
}
}
}
@@ -85,6 +85,12 @@ export const CloudWorkspaceMembersPanel = ({
membersService.members.revalidate();
}, [membersService]);
useEffect(() => {
if (isOwnerOrAdmin) {
workspaceShareSettingService.sharePreview.revalidateInviteLink();
}
}, [isOwnerOrAdmin, workspaceShareSettingService.sharePreview]);
const workspaceQuotaService = useService(WorkspaceQuotaService);
useEffect(() => {
workspaceQuotaService.quota.revalidate();
@@ -178,7 +184,7 @@ export const CloudWorkspaceMembersPanel = ({
const onGenerateInviteLink = useCallback(
async (expireTime: WorkspaceInviteLinkExpireTime) => {
const { link } = await membersService.generateInviteLink(expireTime);
workspaceShareSettingService.sharePreview.revalidate();
workspaceShareSettingService.sharePreview.revalidateInviteLink();
return link;
},
[membersService, workspaceShareSettingService.sharePreview]
@@ -186,7 +192,7 @@ export const CloudWorkspaceMembersPanel = ({
const onRevokeInviteLink = useCallback(async () => {
const success = await membersService.revokeInviteLink();
workspaceShareSettingService.sharePreview.revalidate();
workspaceShareSettingService.sharePreview.revalidateInviteLink();
return success;
}, [membersService, workspaceShareSettingService.sharePreview]);
@@ -52,7 +52,6 @@ export class WorkspaceShareSetting extends Entity {
this.enableAi$.next(value.enableAi);
this.enableSharing$.next(value.enableSharing);
this.enableUrlPreview$.next(value.enableUrlPreview);
this.inviteLink$.next(value.inviteLink);
}
}),
catchErrorInto(this.error$, error => {
@@ -64,6 +63,22 @@ export class WorkspaceShareSetting extends Entity {
})
);
revalidateInviteLink = effect(
exhaustMap(() => {
return fromPromise(signal =>
this.store.fetchInviteLink(this.workspaceService.workspace.id, signal)
).pipe(
smartRetry(),
tap(value => {
this.inviteLink$.next(value);
}),
catchErrorInto(this.error$, error => {
logger.error('Failed to fetch workspace invite link', error);
})
);
})
);
async waitForRevalidation(signal?: AbortSignal) {
this.revalidate();
await this.isLoading$.waitFor(isLoading => !isLoading, signal);
@@ -95,5 +110,6 @@ export class WorkspaceShareSetting extends Entity {
override dispose(): void {
this.revalidate.unsubscribe();
this.revalidateInviteLink.unsubscribe();
}
}
@@ -1,6 +1,7 @@
import type { WorkspaceServerService } from '@affine/core/modules/cloud';
import {
getWorkspaceConfigQuery,
getWorkspaceInviteLinkQuery,
setEnableAiMutation,
setEnableSharingMutation,
setEnableUrlPreviewMutation,
@@ -28,6 +29,22 @@ export class WorkspaceShareSettingStore extends Store {
return data.workspace;
}
async fetchInviteLink(workspaceId: string, signal?: AbortSignal) {
if (!this.workspaceServerService.server) {
throw new Error('No Server');
}
const data = await this.workspaceServerService.server.gql({
query: getWorkspaceInviteLinkQuery,
variables: {
id: workspaceId,
},
context: {
signal,
},
});
return data.workspace.inviteLink;
}
async updateWorkspaceEnableAi(
workspaceId: string,
enableAi: boolean,