Compare commits
8 Commits
8de5736a59
...
feat/triag
| Author | SHA1 | Date | |
|---|---|---|---|
| 7cb046d785 | |||
| e635faea9c | |||
| 30069377e7 | |||
| b1f7467161 | |||
| 7e944a08f6 | |||
| e45ad91340 | |||
| 3e96de730a | |||
| f54ed6abb5 |
@@ -32,6 +32,12 @@ introspects the user/guild tokens Center issues.
|
||||
- **Channel canvas** — one pinned document per channel (`md`/`html`/`text`),
|
||||
re-share replaces, only the original sharer may update/remove; emits
|
||||
`canvas.updated` / `canvas.removed`.
|
||||
- **Slash-command registry** — guild-global catalog: `PUT /api/commands`
|
||||
(the OpenClaw plugin syncs OpenClaw's native-command specs here),
|
||||
`GET /api/commands` (frontend `/` autocomplete). Stored verbatim;
|
||||
execution is unchanged (a `/<cmd>` message flows normally to the plugin →
|
||||
OpenClaw command system; only `/no-reply`,`/force-proceed` are
|
||||
server-intercepted).
|
||||
- **Realtime** — socket.io `/realtime`; `join_channel`/`leave_channel`,
|
||||
`message.created/updated/deleted`, `canvas.*`, presence, typing.
|
||||
|
||||
|
||||
@@ -7,6 +7,7 @@
|
||||
"scripts": {
|
||||
"build": "tsc -p tsconfig.build.json",
|
||||
"start": "node dist/main.js",
|
||||
"print:commands-key": "node dist/cli/print-commands-sync-key.js",
|
||||
"start:dev": "ts-node src/main.ts",
|
||||
"lint": "eslint 'src/**/*.ts'",
|
||||
"lint:fix": "eslint 'src/**/*.ts' --fix",
|
||||
|
||||
@@ -15,6 +15,7 @@ import { RealtimeModule } from './realtime/realtime.module.js';
|
||||
import { MembersModule } from './members/members.module.js';
|
||||
import { FilesModule } from './files/files.module.js';
|
||||
import { CanvasModule } from './canvas/canvas.module.js';
|
||||
import { CommandsModule } from './commands/commands.module.js';
|
||||
|
||||
@Module({
|
||||
imports: [
|
||||
@@ -28,6 +29,7 @@ import { CanvasModule } from './canvas/canvas.module.js';
|
||||
MessagingModule,
|
||||
FilesModule,
|
||||
CanvasModule,
|
||||
CommandsModule,
|
||||
],
|
||||
controllers: [HealthController, MetricsController],
|
||||
providers: [
|
||||
|
||||
@@ -5,8 +5,9 @@ import { Channel } from '../entities/channel.entity.js';
|
||||
import { ChannelMember } from '../entities/channel-member.entity.js';
|
||||
import { WakeMapping } from '../entities/wake-mapping.entity.js';
|
||||
import { TurnService } from './turn.service.js';
|
||||
import { RealtimeGateway } from '../realtime/realtime.gateway.js';
|
||||
|
||||
const X_TYPES = ['general', 'work', 'report', 'discuss', 'triage', 'custom'] as const;
|
||||
const X_TYPES = ['general', 'work', 'report', 'discuss', 'triage', 'custom', 'dm'] as const;
|
||||
type XType = (typeof X_TYPES)[number];
|
||||
|
||||
type CreateChannelInput = {
|
||||
@@ -35,8 +36,34 @@ export class ChannelsService {
|
||||
@InjectRepository(WakeMapping)
|
||||
private readonly wakeRepo: Repository<WakeMapping>,
|
||||
private readonly turnService: TurnService,
|
||||
// RealtimeGateway is provided by the global RealtimeModule. Used to
|
||||
// push channel.joined / channel.left so connected clients (e.g. the
|
||||
// OpenClaw fabric plugin) can sub/unsub socket.io rooms immediately
|
||||
// instead of waiting for the polling fallback.
|
||||
private readonly realtime: RealtimeGateway,
|
||||
) {}
|
||||
|
||||
// Push a channel membership change to each affected user's socket-room.
|
||||
// Best-effort: offline users see the new state on their next connect
|
||||
// (the inbound runs an initial channel-list fetch on connect).
|
||||
private notifyMembership(
|
||||
kind: 'joined' | 'left',
|
||||
channelId: string,
|
||||
userIds: string[] | Set<string>,
|
||||
extra: Record<string, unknown> = {},
|
||||
): void {
|
||||
const ids = userIds instanceof Set ? [...userIds] : userIds;
|
||||
const payload = {
|
||||
channelId,
|
||||
...extra,
|
||||
occurredAt: new Date().toISOString(),
|
||||
};
|
||||
for (const u of ids) {
|
||||
if (!u) continue;
|
||||
this.realtime.emitToUser(u, `channel.${kind}`, { ...payload, userId: u });
|
||||
}
|
||||
}
|
||||
|
||||
// Channels visible to a user within a guild:
|
||||
// - every public channel of the guild (incl. ones created before the user
|
||||
// joined the guild), OR
|
||||
@@ -93,6 +120,7 @@ export class ChannelsService {
|
||||
if (channel.xType === 'discuss' || channel.xType === 'work') {
|
||||
await this.turnService.onMemberAdded(channelId, userId);
|
||||
}
|
||||
this.notifyMembership('joined', channelId, [userId], { xType: channel.xType });
|
||||
}
|
||||
return { status: 'ok', channelId, userId, member: true };
|
||||
}
|
||||
@@ -102,11 +130,14 @@ export class ChannelsService {
|
||||
if (!channel) throw new NotFoundException('channel not found');
|
||||
|
||||
// remove every channel-scoped row that references this user
|
||||
await this.memberRepo.delete({ channelId, userId });
|
||||
const deleted = await this.memberRepo.delete({ channelId, userId });
|
||||
await this.wakeRepo.delete({ channelId, userId });
|
||||
if (channel.xType === 'discuss' || channel.xType === 'work') {
|
||||
await this.turnService.onMemberRemoved(channelId, userId);
|
||||
}
|
||||
if ((deleted.affected ?? 0) > 0) {
|
||||
this.notifyMembership('left', channelId, [userId], { xType: channel.xType });
|
||||
}
|
||||
return { status: 'ok', channelId, userId, member: false };
|
||||
}
|
||||
|
||||
@@ -130,14 +161,19 @@ export class ChannelsService {
|
||||
.map((x) => String(x ?? '').trim())
|
||||
.filter(Boolean);
|
||||
|
||||
// dm channels are always private (a 1:1 conversation); never public.
|
||||
// dm is not unique — multiple dm channels between the same users are
|
||||
// allowed (create() always makes a fresh one, no dedup).
|
||||
const isPublic = xType === 'dm' ? false : Boolean(input.isPublic);
|
||||
|
||||
const channel = await this.channelRepo.save(
|
||||
this.channelRepo.create({
|
||||
guildId,
|
||||
name,
|
||||
xType,
|
||||
kind: input.kind === 'announcement' ? 'announcement' : 'text',
|
||||
isPrivate: !input.isPublic,
|
||||
isPublic: Boolean(input.isPublic),
|
||||
isPrivate: !isPublic,
|
||||
isPublic,
|
||||
lastSeq: 0,
|
||||
}),
|
||||
);
|
||||
@@ -155,6 +191,12 @@ export class ChannelsService {
|
||||
[...memberIds].map((userId) => this.memberRepo.create({ channelId: channel.id, userId })),
|
||||
);
|
||||
|
||||
// Push channel.joined to every seeded member (creator + invitees +
|
||||
// triage on-duty) so their connected sockets sub the new room
|
||||
// immediately. Skips offline users — next connect's channel-list
|
||||
// fetch covers them.
|
||||
this.notifyMembership('joined', channel.id, memberIds, { xType });
|
||||
|
||||
// wake_mapping: triage -> the on-duty user; custom -> each listener
|
||||
const wakeUserIds = new Set<string>();
|
||||
if (xType === 'triage') wakeUserIds.add(onDuty);
|
||||
|
||||
39
src/cli/admin-refresh.ts
Normal file
39
src/cli/admin-refresh.ts
Normal file
@@ -0,0 +1,39 @@
|
||||
// Operator convenience: force-refresh the in-memory Center admin cache
|
||||
// without waiting for the 1-day TTL. Used after `center user set-admin`
|
||||
// to make new admin visible immediately to triage delivery.
|
||||
//
|
||||
// Usage (inside the deployed container):
|
||||
// docker exec fabric-backend-guild node dist/cli/admin-refresh.js
|
||||
//
|
||||
// Prints the (possibly null) result as JSON. Exit 0 always — a "no
|
||||
// admin" outcome is a valid state, not an error.
|
||||
|
||||
import 'reflect-metadata';
|
||||
import { NestFactory } from '@nestjs/core';
|
||||
import { AppModule } from '../app.module.js';
|
||||
import { AdminCacheService } from '../common/admin-cache.service.js';
|
||||
|
||||
async function main() {
|
||||
const app = await NestFactory.createApplicationContext(AppModule, { logger: ['error', 'warn'] });
|
||||
try {
|
||||
const cache = app.get(AdminCacheService);
|
||||
const before = cache.snapshot();
|
||||
const after = await cache.get(true);
|
||||
process.stdout.write(
|
||||
JSON.stringify({
|
||||
ok: true,
|
||||
before,
|
||||
after,
|
||||
changed: JSON.stringify(before) !== JSON.stringify(after),
|
||||
}) + '\n',
|
||||
);
|
||||
} finally {
|
||||
await app.close();
|
||||
}
|
||||
}
|
||||
|
||||
void main().catch((error: unknown) => {
|
||||
const message = error instanceof Error ? error.message : 'unknown error';
|
||||
process.stderr.write(JSON.stringify({ ok: false, error: message }) + '\n');
|
||||
process.exit(1);
|
||||
});
|
||||
37
src/cli/print-commands-sync-key.ts
Normal file
37
src/cli/print-commands-sync-key.ts
Normal file
@@ -0,0 +1,37 @@
|
||||
// Operator convenience (Guild C-2): print the commands-sync key that this
|
||||
// guild process actually has in its environment, so it can be copied into
|
||||
// the OpenClaw plugin's FABRIC_COMMANDS_SYNC_KEY.
|
||||
//
|
||||
// Usage (inside the deployed container — authoritative, reflects compose):
|
||||
// docker exec fabric-backend-guild node dist/cli/print-commands-sync-key.js
|
||||
// docker exec fabric-backend-guild node dist/cli/print-commands-sync-key.js --export
|
||||
//
|
||||
// Default: prints the raw value only (so KEY=$(... ) works).
|
||||
// --export: prints `FABRIC_COMMANDS_SYNC_KEY=<value>` for pasting.
|
||||
// Exit 1 (no stdout) when unset — guild is then in the weaker
|
||||
// "any authenticated user" fallback for PUT /commands.
|
||||
|
||||
const args = new Set(process.argv.slice(2));
|
||||
|
||||
if (args.has('--help') || args.has('-h')) {
|
||||
process.stderr.write(
|
||||
'print-commands-sync-key: outputs FABRIC_BACKEND_GUILD_COMMANDS_SYNC_KEY\n' +
|
||||
' (no flag) print the raw key value\n' +
|
||||
' --export print FABRIC_COMMANDS_SYNC_KEY=<value>\n',
|
||||
);
|
||||
process.exit(0);
|
||||
}
|
||||
|
||||
const key = (process.env.FABRIC_BACKEND_GUILD_COMMANDS_SYNC_KEY ?? '').trim();
|
||||
|
||||
if (!key) {
|
||||
process.stderr.write(
|
||||
'FABRIC_BACKEND_GUILD_COMMANDS_SYNC_KEY is not set — PUT /commands is in ' +
|
||||
'the fallback mode (any authenticated user). Set it to harden (Guild C-2).\n',
|
||||
);
|
||||
process.exit(1);
|
||||
}
|
||||
|
||||
process.stdout.write(
|
||||
(args.has('--export') ? `FABRIC_COMMANDS_SYNC_KEY=${key}` : key) + '\n',
|
||||
);
|
||||
57
src/commands/commands.controller.ts
Normal file
57
src/commands/commands.controller.ts
Normal file
@@ -0,0 +1,57 @@
|
||||
import {
|
||||
Body,
|
||||
Controller,
|
||||
ForbiddenException,
|
||||
Get,
|
||||
Headers,
|
||||
Put,
|
||||
Req,
|
||||
UnauthorizedException,
|
||||
} from '@nestjs/common';
|
||||
import { timingSafeEqual } from 'node:crypto';
|
||||
import { CommandsService } from './commands.service.js';
|
||||
import { SyncCommandsDto } from './dto.sync-commands.dto.js';
|
||||
|
||||
type AuthedRequest = { userId?: string };
|
||||
|
||||
function safeEqual(a: string, b: string): boolean {
|
||||
const ab = Buffer.from(a);
|
||||
const bb = Buffer.from(b);
|
||||
if (ab.length !== bb.length) return false;
|
||||
return timingSafeEqual(ab, bb);
|
||||
}
|
||||
|
||||
@Controller('commands')
|
||||
export class CommandsController {
|
||||
constructor(private readonly commands: CommandsService) {}
|
||||
|
||||
// Guild C-2: catalog write is privileged. When
|
||||
// FABRIC_BACKEND_GUILD_COMMANDS_SYNC_KEY is configured (recommended in
|
||||
// production), the caller must present a matching x-commands-sync-key
|
||||
// header — this restricts writes to the OpenClaw plugin. When unset, we
|
||||
// fall back to "any authenticated agent/user" (never weaker than before).
|
||||
// The body is always strictly validated + size-capped via SyncCommandsDto.
|
||||
@Put()
|
||||
sync(
|
||||
@Req() req: AuthedRequest,
|
||||
@Body() body: SyncCommandsDto,
|
||||
@Headers('x-commands-sync-key') syncKey?: string,
|
||||
) {
|
||||
const configured = process.env.FABRIC_BACKEND_GUILD_COMMANDS_SYNC_KEY ?? '';
|
||||
if (configured) {
|
||||
if (!syncKey || !safeEqual(syncKey, configured)) {
|
||||
throw new ForbiddenException('invalid commands sync key');
|
||||
}
|
||||
} else if (!req.userId) {
|
||||
throw new UnauthorizedException('missing user');
|
||||
}
|
||||
return this.commands.sync(body.commands as unknown[]);
|
||||
}
|
||||
|
||||
// Frontend reads the catalog to drive `/` autocomplete.
|
||||
@Get()
|
||||
list(@Req() req: AuthedRequest) {
|
||||
if (!req.userId) throw new UnauthorizedException('missing user');
|
||||
return this.commands.list();
|
||||
}
|
||||
}
|
||||
12
src/commands/commands.module.ts
Normal file
12
src/commands/commands.module.ts
Normal file
@@ -0,0 +1,12 @@
|
||||
import { Module } from '@nestjs/common';
|
||||
import { TypeOrmModule } from '@nestjs/typeorm';
|
||||
import { GuildCommand } from '../entities/guild-command.entity.js';
|
||||
import { CommandsController } from './commands.controller.js';
|
||||
import { CommandsService } from './commands.service.js';
|
||||
|
||||
@Module({
|
||||
imports: [TypeOrmModule.forFeature([GuildCommand])],
|
||||
controllers: [CommandsController],
|
||||
providers: [CommandsService],
|
||||
})
|
||||
export class CommandsModule {}
|
||||
39
src/commands/commands.service.ts
Normal file
39
src/commands/commands.service.ts
Normal file
@@ -0,0 +1,39 @@
|
||||
import { Injectable } from '@nestjs/common';
|
||||
import { InjectRepository } from '@nestjs/typeorm';
|
||||
import { Repository } from 'typeorm';
|
||||
import { GuildCommand } from '../entities/guild-command.entity.js';
|
||||
|
||||
// This node's guild id (one guild per node).
|
||||
function guildId(): string {
|
||||
return process.env.FABRIC_BACKEND_GUILD_NODE_ID ?? 'guild';
|
||||
}
|
||||
|
||||
@Injectable()
|
||||
export class CommandsService {
|
||||
constructor(
|
||||
@InjectRepository(GuildCommand)
|
||||
private readonly repo: Repository<GuildCommand>,
|
||||
) {}
|
||||
|
||||
// Replace the whole guild-global slash-command catalog (idempotent;
|
||||
// the plugin re-PUTs the full set on every gateway start).
|
||||
async sync(commands: unknown[]): Promise<{ status: string; count: number }> {
|
||||
const gid = guildId();
|
||||
let row = await this.repo.findOne({ where: { guildId: gid } });
|
||||
if (row) {
|
||||
row.commands = commands;
|
||||
} else {
|
||||
row = this.repo.create({ guildId: gid, commands });
|
||||
}
|
||||
await this.repo.save(row);
|
||||
return { status: 'ok', count: commands.length };
|
||||
}
|
||||
|
||||
async list(): Promise<{ commands: unknown[]; updatedAt: string | null }> {
|
||||
const row = await this.repo.findOne({ where: { guildId: guildId() } });
|
||||
return {
|
||||
commands: row?.commands ?? [],
|
||||
updatedAt: row?.updatedAt ? row.updatedAt.toISOString() : null,
|
||||
};
|
||||
}
|
||||
}
|
||||
102
src/commands/dto.sync-commands.dto.ts
Normal file
102
src/commands/dto.sync-commands.dto.ts
Normal file
@@ -0,0 +1,102 @@
|
||||
import {
|
||||
ArrayMaxSize,
|
||||
IsArray,
|
||||
IsBoolean,
|
||||
IsOptional,
|
||||
IsString,
|
||||
MaxLength,
|
||||
ValidateNested,
|
||||
} from 'class-validator';
|
||||
import { Type } from 'class-transformer';
|
||||
|
||||
// Guild C-2: the slash-command catalog is guild-global and rendered by the
|
||||
// frontend `/` autocomplete. Without a strict schema + caps a single
|
||||
// authenticated caller could poison it or blow up the DB / clients.
|
||||
// The global ValidationPipe runs with { whitelist, forbidNonWhitelisted },
|
||||
// so any unknown field is rejected.
|
||||
|
||||
class CommandChoiceDto {
|
||||
@IsString()
|
||||
@MaxLength(200)
|
||||
value!: string;
|
||||
|
||||
@IsString()
|
||||
@MaxLength(200)
|
||||
label!: string;
|
||||
}
|
||||
|
||||
class CommandArgDto {
|
||||
@IsString()
|
||||
@MaxLength(100)
|
||||
name!: string;
|
||||
|
||||
@IsOptional()
|
||||
@IsString()
|
||||
@MaxLength(500)
|
||||
description?: string;
|
||||
|
||||
@IsOptional()
|
||||
@IsString()
|
||||
@MaxLength(40)
|
||||
type?: string;
|
||||
|
||||
@IsOptional()
|
||||
@IsBoolean()
|
||||
required?: boolean;
|
||||
|
||||
@IsOptional()
|
||||
@IsBoolean()
|
||||
captureRemaining?: boolean;
|
||||
|
||||
@IsOptional()
|
||||
@IsBoolean()
|
||||
preferAutocomplete?: boolean;
|
||||
|
||||
// null when there are no choices (plugin sends explicit null).
|
||||
@IsOptional()
|
||||
@IsArray()
|
||||
@ArrayMaxSize(100)
|
||||
@ValidateNested({ each: true })
|
||||
@Type(() => CommandChoiceDto)
|
||||
choices?: CommandChoiceDto[] | null;
|
||||
}
|
||||
|
||||
class CommandSpecDto {
|
||||
@IsString()
|
||||
@MaxLength(100)
|
||||
name!: string;
|
||||
|
||||
@IsOptional()
|
||||
@IsString()
|
||||
@MaxLength(100)
|
||||
nativeName?: string;
|
||||
|
||||
@IsOptional()
|
||||
@IsString()
|
||||
@MaxLength(500)
|
||||
description?: string;
|
||||
|
||||
@IsOptional()
|
||||
@IsBoolean()
|
||||
acceptsArgs?: boolean;
|
||||
|
||||
@IsOptional()
|
||||
@IsArray()
|
||||
@ArrayMaxSize(50)
|
||||
@ValidateNested({ each: true })
|
||||
@Type(() => CommandArgDto)
|
||||
args?: CommandArgDto[];
|
||||
|
||||
@IsOptional()
|
||||
@IsString()
|
||||
@MaxLength(20)
|
||||
argsParsing?: string;
|
||||
}
|
||||
|
||||
export class SyncCommandsDto {
|
||||
@IsArray()
|
||||
@ArrayMaxSize(200)
|
||||
@ValidateNested({ each: true })
|
||||
@Type(() => CommandSpecDto)
|
||||
commands!: CommandSpecDto[];
|
||||
}
|
||||
73
src/common/admin-cache.service.ts
Normal file
73
src/common/admin-cache.service.ts
Normal file
@@ -0,0 +1,73 @@
|
||||
/**
|
||||
* Center-scoped admin cache.
|
||||
*
|
||||
* Holds the at-most-one admin user (email + userId) fetched from Center.
|
||||
* Used to decide who to deliver triage messages to as a silent observer
|
||||
* (wake=false), regardless of on-duty / mention status.
|
||||
*
|
||||
* Refresh policy (per spec, 2026-05-22):
|
||||
* • TTL = 1 day. Center admin changes are rare; agents tolerate a
|
||||
* day's stale cache without surprises
|
||||
* • on first lookup the cache lazy-fetches
|
||||
* • cli `admin refresh` forces an out-of-band refresh without waiting
|
||||
* for TTL expiry
|
||||
*
|
||||
* Failure mode: a Center fetch error is treated identically to "no
|
||||
* admin" — guild keeps operating without an observer. The cache holds
|
||||
* the failed-fetch decision for the same TTL so we don't hammer Center.
|
||||
*/
|
||||
|
||||
import { Injectable, Logger } from '@nestjs/common';
|
||||
import { fetchAdminEmail } from './center-auth.js';
|
||||
|
||||
const ADMIN_CACHE_TTL_MS = 24 * 60 * 60 * 1000;
|
||||
|
||||
export interface CachedAdmin {
|
||||
email: string;
|
||||
userId: string;
|
||||
}
|
||||
|
||||
@Injectable()
|
||||
export class AdminCacheService {
|
||||
private readonly logger = new Logger(AdminCacheService.name);
|
||||
private cached: CachedAdmin | null = null;
|
||||
private cachedAt = 0;
|
||||
private inflight: Promise<CachedAdmin | null> | null = null;
|
||||
|
||||
/**
|
||||
* Return the cached admin, fetching from Center if the cache is empty
|
||||
* or older than the TTL. Returns null if no admin is set.
|
||||
*
|
||||
* `force=true` bypasses the cache and refreshes immediately — used by
|
||||
* the cli refresh command.
|
||||
*/
|
||||
async get(force = false): Promise<CachedAdmin | null> {
|
||||
const fresh = Date.now() - this.cachedAt < ADMIN_CACHE_TTL_MS;
|
||||
if (!force && this.cachedAt > 0 && fresh) {
|
||||
return this.cached;
|
||||
}
|
||||
if (this.inflight) return this.inflight;
|
||||
|
||||
this.inflight = (async () => {
|
||||
try {
|
||||
const result = await fetchAdminEmail();
|
||||
this.cached = result;
|
||||
this.cachedAt = Date.now();
|
||||
this.logger.log(
|
||||
`admin cache refreshed: ${result ? `${result.email} (${result.userId})` : 'no admin set'}`,
|
||||
);
|
||||
return result;
|
||||
} finally {
|
||||
this.inflight = null;
|
||||
}
|
||||
})();
|
||||
return this.inflight;
|
||||
}
|
||||
|
||||
/** Snapshot of the cached admin (no fetch). Returns null if not yet
|
||||
* populated. Used by the hot delivery path which doesn't want to
|
||||
* block on a Center round-trip. */
|
||||
snapshot(): CachedAdmin | null {
|
||||
return this.cached;
|
||||
}
|
||||
}
|
||||
@@ -26,6 +26,31 @@ export async function introspectGuildToken(token: string): Promise<{ active: boo
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Fetch the single Center-scoped admin user (if any).
|
||||
* Same x-api-key auth as introspect / resolve-names.
|
||||
* Returns `null` when no admin is set OR the request fails (treated
|
||||
* identically — the guild simply falls back to "no admin observer").
|
||||
*/
|
||||
export async function fetchAdminEmail(): Promise<{ email: string; userId: string } | null> {
|
||||
const centerBaseUrl = process.env.FABRIC_BACKEND_GUILD_CENTER_BASE_URL;
|
||||
const centerApiKey = process.env.FABRIC_BACKEND_GUILD_CENTER_API_KEY;
|
||||
if (!centerBaseUrl || !centerApiKey) return null;
|
||||
|
||||
try {
|
||||
const res = await fetch(`${centerBaseUrl}/api/auth/admin-email`, {
|
||||
method: 'GET',
|
||||
headers: { 'x-api-key': centerApiKey },
|
||||
});
|
||||
if (!res.ok) return null;
|
||||
const data = (await res.json()) as { email?: string; userId?: string } | null;
|
||||
if (!data || !data.email || !data.userId) return null;
|
||||
return { email: data.email, userId: data.userId };
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
// Resolve <@user.name:NAME> names to userIds within this guild node via
|
||||
// Center. Unresolved names are simply absent from the returned map.
|
||||
export async function resolveUserNames(names: string[]): Promise<Record<string, string>> {
|
||||
|
||||
@@ -13,6 +13,7 @@ import { GuildMemberRole } from './entities/guild-member-role.entity.js';
|
||||
import { IdempotencyRecord } from './entities/idempotency-record.entity.js';
|
||||
import { StoredFile } from './entities/stored-file.entity.js';
|
||||
import { ChannelCanvas } from './entities/channel-canvas.entity.js';
|
||||
import { GuildCommand } from './entities/guild-command.entity.js';
|
||||
|
||||
export const buildTypeOrmConfig = (): TypeOrmModuleOptions => ({
|
||||
type: 'mysql',
|
||||
@@ -36,6 +37,7 @@ export const buildTypeOrmConfig = (): TypeOrmModuleOptions => ({
|
||||
IdempotencyRecord,
|
||||
StoredFile,
|
||||
ChannelCanvas,
|
||||
GuildCommand,
|
||||
],
|
||||
synchronize: (process.env.FABRIC_BACKEND_GUILD_DB_SYNC ?? 'true') === 'true',
|
||||
logging: (process.env.FABRIC_BACKEND_GUILD_DB_LOGGING ?? 'false') === 'true',
|
||||
|
||||
@@ -16,9 +16,9 @@ export class Channel {
|
||||
@Column({
|
||||
name: 'x_type',
|
||||
type: 'enum',
|
||||
enum: ['general', 'work', 'report', 'discuss', 'triage', 'custom'],
|
||||
enum: ['general', 'work', 'report', 'discuss', 'triage', 'custom', 'dm'],
|
||||
})
|
||||
xType!: 'general' | 'work' | 'report' | 'discuss' | 'triage' | 'custom';
|
||||
xType!: 'general' | 'work' | 'report' | 'discuss' | 'triage' | 'custom' | 'dm';
|
||||
|
||||
@Column({ type: 'varchar', length: 16, default: 'text' })
|
||||
kind!: 'text' | 'announcement';
|
||||
|
||||
25
src/entities/guild-command.entity.ts
Normal file
25
src/entities/guild-command.entity.ts
Normal file
@@ -0,0 +1,25 @@
|
||||
import { Column, Entity, Index, PrimaryGeneratedColumn, UpdateDateColumn } from 'typeorm';
|
||||
|
||||
// Guild-global slash-command catalog. One row per guild (this node's
|
||||
// FABRIC_BACKEND_GUILD_NODE_ID). The OpenClaw plugin PUTs the OpenClaw
|
||||
// native-command specs here (the same data Discord registers as slash
|
||||
// commands); the frontend GETs it to drive `/` autocomplete. The guild
|
||||
// node stores the catalog opaquely — it does not interpret command bodies.
|
||||
@Entity('guild_commands')
|
||||
export class GuildCommand {
|
||||
@PrimaryGeneratedColumn('uuid')
|
||||
id!: string;
|
||||
|
||||
@Index({ unique: true })
|
||||
@Column({ name: 'guild_id', type: 'varchar', length: 80 })
|
||||
guildId!: string;
|
||||
|
||||
// NativeCommandSpec[]-shaped (name, nativeName, description, acceptsArgs,
|
||||
// args[{name,description,type,required,choices:[{value,label}],
|
||||
// captureRemaining,preferAutocomplete}], argsParsing). Stored verbatim.
|
||||
@Column({ type: 'json' })
|
||||
commands!: unknown[];
|
||||
|
||||
@UpdateDateColumn()
|
||||
updatedAt!: Date;
|
||||
}
|
||||
@@ -3,6 +3,7 @@ import {
|
||||
ConflictException,
|
||||
Controller,
|
||||
Delete,
|
||||
ForbiddenException,
|
||||
Get,
|
||||
Headers,
|
||||
NotFoundException,
|
||||
@@ -10,14 +11,17 @@ import {
|
||||
Patch,
|
||||
Post,
|
||||
Query,
|
||||
Req,
|
||||
} from '@nestjs/common';
|
||||
import { InjectRepository } from '@nestjs/typeorm';
|
||||
import { DataSource, Repository } from 'typeorm';
|
||||
import { CreateMessageDto } from './dto.create-message.dto.js';
|
||||
import { Channel } from '../entities/channel.entity.js';
|
||||
import { ChannelMember } from '../entities/channel-member.entity.js';
|
||||
import { Message } from '../entities/message.entity.js';
|
||||
import { IdempotencyRecord } from '../entities/idempotency-record.entity.js';
|
||||
import { WakeMapping } from '../entities/wake-mapping.entity.js';
|
||||
import { AdminCacheService } from '../common/admin-cache.service.js';
|
||||
import { parseSlashCommand } from '../channels/slash-commands.js';
|
||||
import { parseMentions, extractNameMentions, replaceNameMentions } from '../channels/mentions.js';
|
||||
import { resolveUserNames } from '../common/center-auth.js';
|
||||
@@ -36,6 +40,8 @@ export class MessagingController {
|
||||
private readonly dataSource: DataSource,
|
||||
@InjectRepository(Channel)
|
||||
private readonly channelRepo: Repository<Channel>,
|
||||
@InjectRepository(ChannelMember)
|
||||
private readonly memberRepo: Repository<ChannelMember>,
|
||||
@InjectRepository(Message)
|
||||
private readonly messageRepo: Repository<Message>,
|
||||
@InjectRepository(IdempotencyRecord)
|
||||
@@ -45,6 +51,7 @@ export class MessagingController {
|
||||
private readonly turn: TurnService,
|
||||
private readonly events: EventsService,
|
||||
private readonly realtime: RealtimeGateway,
|
||||
private readonly adminCache: AdminCacheService,
|
||||
) {}
|
||||
|
||||
private async getIdempotentResponse(
|
||||
@@ -86,6 +93,19 @@ export class MessagingController {
|
||||
};
|
||||
}
|
||||
|
||||
// Channel-participant gate (Guild C-1): public channels are readable/
|
||||
// writable by any authenticated user; private channels require explicit
|
||||
// channel_members membership. Returns the channel so callers can reuse it.
|
||||
private async assertParticipant(channelId: string, userId: string): Promise<Channel> {
|
||||
const channel = await this.channelRepo.findOne({ where: { id: channelId } });
|
||||
if (!channel) throw new NotFoundException('channel not found');
|
||||
if (channel.isPublic) return channel;
|
||||
if (!userId) throw new ForbiddenException('not a channel member');
|
||||
const member = await this.memberRepo.findOne({ where: { channelId, userId } });
|
||||
if (!member) throw new ForbiddenException('not a channel member');
|
||||
return channel;
|
||||
}
|
||||
|
||||
// Persists one message (allocates a seq under a channel row lock) and
|
||||
// returns its view. Used for normal messages and for guild /ack messages.
|
||||
private async persistMessage(
|
||||
@@ -136,20 +156,25 @@ export class MessagingController {
|
||||
async create(
|
||||
@Param('id') channelId: string,
|
||||
@Body() body: CreateMessageDto,
|
||||
@Req() req: { userId?: string },
|
||||
@Headers('idempotency-key') idempotencyKey?: string,
|
||||
) {
|
||||
const scope = `POST:/channels/${channelId}/messages`;
|
||||
const existed = await this.getIdempotentResponse(scope, idempotencyKey);
|
||||
if (existed) return existed;
|
||||
|
||||
const channel = await this.channelRepo.findOne({ where: { id: channelId } });
|
||||
if (!channel) throw new NotFoundException('channel not found');
|
||||
// Guild C-1: caller must be a participant of the channel, and the
|
||||
// author is always the authenticated user — body.authorUserId is
|
||||
// ignored so a caller can never post as someone else.
|
||||
const userId = String(req.userId ?? '');
|
||||
if (!userId) throw new ForbiddenException('missing user');
|
||||
const channel = await this.assertParticipant(channelId, userId);
|
||||
if (channel.closed) {
|
||||
throw new ConflictException({ error: 'channel_closed', message: 'channel is closed' });
|
||||
}
|
||||
const xType = channel.xType ?? 'general';
|
||||
const isRotating = xType === 'discuss' || xType === 'work';
|
||||
const authorUserId = String(body.authorUserId ?? 'anonymous');
|
||||
const authorUserId = userId;
|
||||
|
||||
// ---- translate <@user.name:NAME> -> <@userId> (outside backticks) via
|
||||
// Center before anything else persists/parses the content
|
||||
@@ -202,16 +227,19 @@ export class MessagingController {
|
||||
const decision = await this.turn.onNormalMessage(channelId, authorUserId, mentionIds);
|
||||
await this.realtime.emitMessageTargeted(channelId, responseBody, decision.wakeupUserId);
|
||||
} else {
|
||||
// general/report/triage/custom: wakeup from x_type + wake_mapping;
|
||||
// general also honors the message's at-list
|
||||
// general/report/triage/custom: 3-state delivery
|
||||
// (wake / observer / skip) — see realtime.gateway.computeDelivery.
|
||||
// Center-scoped admin (cached, 1d TTL) gets `observer` on triage.
|
||||
const wakeRows = await this.wakeRepo.find({ where: { channelId } });
|
||||
const wakeUserIds = new Set(wakeRows.map((w) => w.userId));
|
||||
const mentionUserIds = new Set(mentionIds.filter((id) => id !== authorUserId));
|
||||
const admin = await this.adminCache.get();
|
||||
await this.realtime.emitMessageCreated(channelId, responseBody, {
|
||||
xType,
|
||||
authorUserId,
|
||||
wakeUserIds,
|
||||
mentionUserIds,
|
||||
adminUserId: admin?.userId ?? null,
|
||||
});
|
||||
}
|
||||
|
||||
@@ -223,14 +251,23 @@ export class MessagingController {
|
||||
@Param('id') channelId: string,
|
||||
@Param('messageId') messageId: string,
|
||||
@Body() body: { content?: string },
|
||||
@Req() req: { userId?: string },
|
||||
@Headers('idempotency-key') idempotencyKey?: string,
|
||||
) {
|
||||
const scope = `PATCH:/channels/${channelId}/messages/${messageId}`;
|
||||
const existed = await this.getIdempotentResponse(scope, idempotencyKey);
|
||||
if (existed) return existed;
|
||||
|
||||
// Guild C-1: participant + author-ownership.
|
||||
const userId = String(req.userId ?? '');
|
||||
if (!userId) throw new ForbiddenException('missing user');
|
||||
await this.assertParticipant(channelId, userId);
|
||||
|
||||
const item = await this.messageRepo.findOne({ where: { channelId, messageId } });
|
||||
if (!item) return { status: 'not_found' };
|
||||
if (item.authorUserId !== userId) {
|
||||
throw new ForbiddenException('not the message author');
|
||||
}
|
||||
|
||||
const now = Date.now();
|
||||
const createdAt = new Date(item.createdAt).getTime();
|
||||
@@ -259,14 +296,23 @@ export class MessagingController {
|
||||
async remove(
|
||||
@Param('id') channelId: string,
|
||||
@Param('messageId') messageId: string,
|
||||
@Req() req: { userId?: string },
|
||||
@Headers('idempotency-key') idempotencyKey?: string,
|
||||
) {
|
||||
const scope = `DELETE:/channels/${channelId}/messages/${messageId}`;
|
||||
const existed = await this.getIdempotentResponse(scope, idempotencyKey);
|
||||
if (existed) return existed;
|
||||
|
||||
// Guild C-1: participant + author-ownership.
|
||||
const userId = String(req.userId ?? '');
|
||||
if (!userId) throw new ForbiddenException('missing user');
|
||||
await this.assertParticipant(channelId, userId);
|
||||
|
||||
const item = await this.messageRepo.findOne({ where: { channelId, messageId } });
|
||||
if (!item) return { status: 'not_found' };
|
||||
if (item.authorUserId !== userId) {
|
||||
throw new ForbiddenException('not the message author');
|
||||
}
|
||||
|
||||
item.isDeleted = true;
|
||||
item.deletedAt = new Date();
|
||||
@@ -304,10 +350,14 @@ export class MessagingController {
|
||||
@Get()
|
||||
async listBySeq(
|
||||
@Param('id') channelId: string,
|
||||
@Req() req: { userId?: string },
|
||||
@Query('seq_from') seqFrom?: string,
|
||||
@Query('seq_to') seqTo?: string,
|
||||
@Query('limit') limit?: string,
|
||||
) {
|
||||
// Guild C-1: only participants may read channel history.
|
||||
const userId = String(req.userId ?? '');
|
||||
if (!userId) throw new ForbiddenException('missing user');
|
||||
const from = seqFrom ? Number(seqFrom) : 1;
|
||||
const to = seqTo ? Number(seqTo) : Number.MAX_SAFE_INTEGER;
|
||||
const safeLimit = clampLimit(limit, DEFAULT_PAGE_LIMIT, MAX_PAGE_LIMIT);
|
||||
@@ -327,10 +377,7 @@ export class MessagingController {
|
||||
};
|
||||
}
|
||||
|
||||
const channel = await this.channelRepo.findOne({ where: { id: channelId } });
|
||||
if (!channel) {
|
||||
throw new NotFoundException('channel not found');
|
||||
}
|
||||
const channel = await this.assertParticipant(channelId, userId);
|
||||
|
||||
const qb = this.messageRepo
|
||||
.createQueryBuilder('m')
|
||||
|
||||
@@ -2,12 +2,16 @@ import { Module } from '@nestjs/common';
|
||||
import { TypeOrmModule } from '@nestjs/typeorm';
|
||||
import { MessagingController } from './messaging.controller.js';
|
||||
import { Channel } from '../entities/channel.entity.js';
|
||||
import { ChannelMember } from '../entities/channel-member.entity.js';
|
||||
import { Message } from '../entities/message.entity.js';
|
||||
import { IdempotencyRecord } from '../entities/idempotency-record.entity.js';
|
||||
import { WakeMapping } from '../entities/wake-mapping.entity.js';
|
||||
import { AdminCacheService } from '../common/admin-cache.service.js';
|
||||
|
||||
@Module({
|
||||
imports: [TypeOrmModule.forFeature([Channel, Message, IdempotencyRecord, WakeMapping])],
|
||||
imports: [TypeOrmModule.forFeature([Channel, ChannelMember, Message, IdempotencyRecord, WakeMapping])],
|
||||
controllers: [MessagingController],
|
||||
providers: [AdminCacheService],
|
||||
exports: [AdminCacheService],
|
||||
})
|
||||
export class MessagingModule {}
|
||||
|
||||
@@ -11,17 +11,72 @@ import { Logger } from '@nestjs/common';
|
||||
import { Server, Socket } from 'socket.io';
|
||||
import { introspectGuildToken } from '../common/center-auth.js';
|
||||
|
||||
type XType = 'general' | 'work' | 'report' | 'discuss' | 'triage' | 'custom';
|
||||
type XType = 'general' | 'work' | 'report' | 'discuss' | 'triage' | 'custom' | 'dm';
|
||||
|
||||
// Wakeup for non-rotating channels only (general/report/triage/custom).
|
||||
// discuss/work go through TurnService + emitMessageTargeted, never here.
|
||||
// Precedence:
|
||||
// 1. the author never gets woken by their own message
|
||||
// 2. triage/custom: only wake users in the channel's wake_mapping
|
||||
// (mentions change nothing here)
|
||||
// 3. general: if the message has an at-list, wake only the at'd users;
|
||||
// otherwise wake everyone
|
||||
// 4. report (and anything else): wake nobody
|
||||
/**
|
||||
* Per-recipient delivery decision for a non-rotating channel message.
|
||||
*
|
||||
* • `wake` — push the event AND wake the recipient (model turn fires)
|
||||
* • `observer` — push the event with wakeup=false (silent; UI displays
|
||||
* but the openclaw plugin records-only without dispatch). Currently
|
||||
* used for the Center admin observing triage traffic
|
||||
* • `skip` — don't even emit the event to this recipient
|
||||
*
|
||||
* Wakeup-only channels (general/report/dm/custom) never return
|
||||
* 'observer'; the legacy behaviour is preserved end-to-end.
|
||||
*
|
||||
* Precedence for triage (the only place 'skip' / 'observer' fire):
|
||||
* 1. author never gets back their own message
|
||||
* 2. wake_mapping (on-duty) → wake
|
||||
* 3. mention → wake (NEW: was 'skip' before — see Fabric PR 'triage
|
||||
* mention exception')
|
||||
* 4. admin (Center-scoped, at most one) → observer
|
||||
* 5. everyone else → skip (was 'deliver, wakeup=false' before)
|
||||
*/
|
||||
export type DeliveryDecision = 'wake' | 'observer' | 'skip';
|
||||
|
||||
export interface ComputeDeliveryArgs {
|
||||
xType: XType;
|
||||
recipientUserId: string;
|
||||
authorUserId: string;
|
||||
wakeUserIds: Set<string>;
|
||||
mentionUserIds?: Set<string>;
|
||||
/** Single Center-scoped admin userId, or null. */
|
||||
adminUserId?: string | null;
|
||||
}
|
||||
|
||||
export function computeDelivery(args: ComputeDeliveryArgs): DeliveryDecision {
|
||||
const { xType, recipientUserId, authorUserId, wakeUserIds, mentionUserIds, adminUserId } = args;
|
||||
if (recipientUserId === authorUserId) return 'skip';
|
||||
|
||||
switch (xType) {
|
||||
case 'triage':
|
||||
if (wakeUserIds.has(recipientUserId)) return 'wake';
|
||||
if (mentionUserIds?.has(recipientUserId)) return 'wake';
|
||||
if (adminUserId && recipientUserId === adminUserId) return 'observer';
|
||||
return 'skip';
|
||||
case 'general':
|
||||
if (mentionUserIds && mentionUserIds.size > 0) {
|
||||
return mentionUserIds.has(recipientUserId) ? 'wake' : 'observer';
|
||||
}
|
||||
return 'wake';
|
||||
case 'custom':
|
||||
// wake_mapping decides who wakes; everyone else still sees the
|
||||
// message (observer) — preserves the legacy "deliver to all, wake
|
||||
// some" contract for custom channels.
|
||||
return wakeUserIds.has(recipientUserId) ? 'wake' : 'observer';
|
||||
case 'dm':
|
||||
return 'wake';
|
||||
default:
|
||||
// report (and anything else): deliver as observer, no wake
|
||||
return 'observer';
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @deprecated Use computeDelivery (returns 3-state). Kept for any
|
||||
* external callers; treats 'observer' and 'skip' both as `false`.
|
||||
*/
|
||||
export function computeWakeup(args: {
|
||||
xType: XType;
|
||||
recipientUserId: string;
|
||||
@@ -29,20 +84,7 @@ export function computeWakeup(args: {
|
||||
wakeUserIds: Set<string>;
|
||||
mentionUserIds?: Set<string>;
|
||||
}): boolean {
|
||||
const { xType, recipientUserId, authorUserId, wakeUserIds, mentionUserIds } = args;
|
||||
if (recipientUserId === authorUserId) return false;
|
||||
switch (xType) {
|
||||
case 'general':
|
||||
if (mentionUserIds && mentionUserIds.size > 0) {
|
||||
return mentionUserIds.has(recipientUserId);
|
||||
}
|
||||
return true;
|
||||
case 'triage':
|
||||
case 'custom':
|
||||
return wakeUserIds.has(recipientUserId);
|
||||
default:
|
||||
return false;
|
||||
}
|
||||
return computeDelivery(args) === 'wake';
|
||||
}
|
||||
|
||||
@WebSocketGateway({
|
||||
@@ -93,6 +135,10 @@ export class RealtimeGateway implements OnGatewayConnection, OnGatewayDisconnect
|
||||
const userId = result.user.id || this.userIdFromClient(client);
|
||||
client.data.userId = userId;
|
||||
this.onlineUsers.add(userId);
|
||||
// Per-user room: lets server code emit user-scoped events (e.g.
|
||||
// channel.joined when membership changes) without bookkeeping a
|
||||
// userId→sockets map. All of this user's sockets receive the event.
|
||||
client.join(`user:${userId}`);
|
||||
this.server.emit('presence.online', {
|
||||
userId,
|
||||
onlineCount: this.onlineUsers.size,
|
||||
@@ -168,7 +214,18 @@ export class RealtimeGateway implements OnGatewayConnection, OnGatewayDisconnect
|
||||
this.server.to(`channel:${channelId}`).emit(event, data);
|
||||
}
|
||||
|
||||
// Emits message.created per-recipient so each carries its own `wakeup` flag.
|
||||
// Emit a user-scoped event to all sockets currently connected for `userId`
|
||||
// (via the `user:<userId>` room joined in handleConnection). No-op for
|
||||
// offline users — the next connect's initial channel-list fetch covers it.
|
||||
emitToUser(userId: string, event: string, data: Record<string, unknown>): void {
|
||||
if (!userId) return;
|
||||
this.server.to(`user:${userId}`).emit(event, data);
|
||||
}
|
||||
|
||||
// Emits message.created per-recipient using the 3-state delivery
|
||||
// decision (wake / observer / skip). Skipped recipients receive
|
||||
// nothing — used by triage channels to keep non-on-duty / non-mention
|
||||
// / non-admin users completely out of the loop.
|
||||
async emitMessageCreated(
|
||||
channelId: string,
|
||||
data: Record<string, unknown>,
|
||||
@@ -177,19 +234,28 @@ export class RealtimeGateway implements OnGatewayConnection, OnGatewayDisconnect
|
||||
authorUserId: string;
|
||||
wakeUserIds: Set<string>;
|
||||
mentionUserIds?: Set<string>;
|
||||
/** Single Center-scoped admin userId (or null). */
|
||||
adminUserId?: string | null;
|
||||
},
|
||||
): Promise<void> {
|
||||
const sockets = await this.server.in(`channel:${channelId}`).fetchSockets();
|
||||
for (const s of sockets) {
|
||||
const recipientUserId = typeof s.data.userId === 'string' ? s.data.userId : `anon:${s.id}`;
|
||||
const wakeup = computeWakeup({
|
||||
const decision = computeDelivery({
|
||||
xType: ctx.xType,
|
||||
recipientUserId,
|
||||
authorUserId: ctx.authorUserId,
|
||||
wakeUserIds: ctx.wakeUserIds,
|
||||
mentionUserIds: ctx.mentionUserIds,
|
||||
adminUserId: ctx.adminUserId,
|
||||
});
|
||||
if (decision === 'skip') continue;
|
||||
s.emit('message.created', {
|
||||
...data,
|
||||
channelId,
|
||||
wakeup: decision === 'wake',
|
||||
xType: ctx.xType,
|
||||
});
|
||||
s.emit('message.created', { ...data, channelId, wakeup });
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user