Compare commits
8 Commits
b1f7467161
...
ca20df7618
| Author | SHA1 | Date | |
|---|---|---|---|
| ca20df7618 | |||
| cb7b3bb5fe | |||
| 985b06a886 | |||
| 80ee9082f3 | |||
| 801b562999 | |||
| 7cb046d785 | |||
| e635faea9c | |||
| 30069377e7 |
42
src/agents/agent-presence.controller.ts
Normal file
42
src/agents/agent-presence.controller.ts
Normal file
@@ -0,0 +1,42 @@
|
|||||||
|
import { BadRequestException, Body, Controller, Get, Param, Put } from '@nestjs/common';
|
||||||
|
import { AgentPresenceService, PresenceStatus } from './agent-presence.service.js';
|
||||||
|
|
||||||
|
const VALID: PresenceStatus[] = ['idle', 'on_call', 'busy', 'exhausted', 'offline', 'unknown'];
|
||||||
|
|
||||||
|
interface PutBody {
|
||||||
|
status?: string;
|
||||||
|
source?: string;
|
||||||
|
}
|
||||||
|
|
||||||
|
@Controller('agents/:userId/presence')
|
||||||
|
export class AgentPresenceController {
|
||||||
|
constructor(private readonly svc: AgentPresenceService) {}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Read a user's current presence cache row.
|
||||||
|
* Auth: ApiKeyGuard (global). Any introspected center user can read.
|
||||||
|
*/
|
||||||
|
@Get()
|
||||||
|
async get(@Param('userId') userId: string): Promise<{ userId: string; status: PresenceStatus }> {
|
||||||
|
const status = await this.svc.getStatus(userId);
|
||||||
|
return { userId, status };
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Push a presence update. Called by Fabric.OpenclawPlugin's
|
||||||
|
* `presence-sync` loop on each delta. Auth: ApiKeyGuard (global) +
|
||||||
|
* the plugin uses its center-introspected api key.
|
||||||
|
*
|
||||||
|
* `source` is a debug tag describing who pushed (e.g. 'hf-plugin',
|
||||||
|
* 'manual'). Stored verbatim for trail.
|
||||||
|
*/
|
||||||
|
@Put()
|
||||||
|
async put(@Param('userId') userId: string, @Body() body: PutBody): Promise<{ userId: string; status: PresenceStatus }> {
|
||||||
|
if (!body?.status || !VALID.includes(body.status as PresenceStatus)) {
|
||||||
|
throw new BadRequestException(`status must be one of ${VALID.join('|')}`);
|
||||||
|
}
|
||||||
|
const source = (body.source ?? 'unknown').slice(0, 64);
|
||||||
|
const row = await this.svc.setStatus(userId, body.status as PresenceStatus, source);
|
||||||
|
return { userId: row.userId, status: row.status };
|
||||||
|
}
|
||||||
|
}
|
||||||
13
src/agents/agent-presence.module.ts
Normal file
13
src/agents/agent-presence.module.ts
Normal file
@@ -0,0 +1,13 @@
|
|||||||
|
import { Module } from '@nestjs/common';
|
||||||
|
import { TypeOrmModule } from '@nestjs/typeorm';
|
||||||
|
import { AgentPresence } from '../entities/agent-presence.entity.js';
|
||||||
|
import { AgentPresenceController } from './agent-presence.controller.js';
|
||||||
|
import { AgentPresenceService } from './agent-presence.service.js';
|
||||||
|
|
||||||
|
@Module({
|
||||||
|
imports: [TypeOrmModule.forFeature([AgentPresence])],
|
||||||
|
controllers: [AgentPresenceController],
|
||||||
|
providers: [AgentPresenceService],
|
||||||
|
exports: [AgentPresenceService],
|
||||||
|
})
|
||||||
|
export class AgentPresenceModule {}
|
||||||
53
src/agents/agent-presence.service.ts
Normal file
53
src/agents/agent-presence.service.ts
Normal file
@@ -0,0 +1,53 @@
|
|||||||
|
import { Injectable } from '@nestjs/common';
|
||||||
|
import { InjectRepository } from '@nestjs/typeorm';
|
||||||
|
import { Repository } from 'typeorm';
|
||||||
|
import { AgentPresence } from '../entities/agent-presence.entity.js';
|
||||||
|
|
||||||
|
export type PresenceStatus = 'idle' | 'on_call' | 'busy' | 'exhausted' | 'offline' | 'unknown';
|
||||||
|
|
||||||
|
@Injectable()
|
||||||
|
export class AgentPresenceService {
|
||||||
|
constructor(
|
||||||
|
@InjectRepository(AgentPresence)
|
||||||
|
private readonly repo: Repository<AgentPresence>,
|
||||||
|
) {}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Get a user's current presence. Returns 'unknown' if no row.
|
||||||
|
* Used by `RealtimeGateway` per-recipient when xType === 'announce'.
|
||||||
|
*/
|
||||||
|
async getStatus(userId: string): Promise<PresenceStatus> {
|
||||||
|
if (!userId) return 'unknown';
|
||||||
|
const row = await this.repo.findOne({ where: { userId } });
|
||||||
|
return row?.status ?? 'unknown';
|
||||||
|
}
|
||||||
|
|
||||||
|
/** Bulk variant for delivery-time lookups across many recipients in one trip. */
|
||||||
|
async getStatusMap(userIds: string[]): Promise<Map<string, PresenceStatus>> {
|
||||||
|
const out = new Map<string, PresenceStatus>();
|
||||||
|
for (const id of userIds) out.set(id, 'unknown');
|
||||||
|
if (userIds.length === 0) return out;
|
||||||
|
const rows = await this.repo
|
||||||
|
.createQueryBuilder('p')
|
||||||
|
.where('p.userId IN (:...ids)', { ids: userIds })
|
||||||
|
.getMany();
|
||||||
|
for (const r of rows) out.set(r.userId, r.status);
|
||||||
|
return out;
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Upsert a user's presence. Source is a free-text tag for debugging
|
||||||
|
* (e.g. "hf-plugin", "manual", "test"). PUT /agents/:id/presence
|
||||||
|
* calls this; the plugin pushes only on diff so writes are sparse.
|
||||||
|
*/
|
||||||
|
async setStatus(userId: string, status: PresenceStatus, source: string): Promise<AgentPresence> {
|
||||||
|
const existing = await this.repo.findOne({ where: { userId } });
|
||||||
|
if (existing) {
|
||||||
|
existing.status = status;
|
||||||
|
existing.source = source;
|
||||||
|
return this.repo.save(existing);
|
||||||
|
}
|
||||||
|
const row = this.repo.create({ userId, status, source });
|
||||||
|
return this.repo.save(row);
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -16,6 +16,7 @@ import { MembersModule } from './members/members.module.js';
|
|||||||
import { FilesModule } from './files/files.module.js';
|
import { FilesModule } from './files/files.module.js';
|
||||||
import { CanvasModule } from './canvas/canvas.module.js';
|
import { CanvasModule } from './canvas/canvas.module.js';
|
||||||
import { CommandsModule } from './commands/commands.module.js';
|
import { CommandsModule } from './commands/commands.module.js';
|
||||||
|
import { AgentPresenceModule } from './agents/agent-presence.module.js';
|
||||||
|
|
||||||
@Module({
|
@Module({
|
||||||
imports: [
|
imports: [
|
||||||
@@ -30,6 +31,7 @@ import { CommandsModule } from './commands/commands.module.js';
|
|||||||
FilesModule,
|
FilesModule,
|
||||||
CanvasModule,
|
CanvasModule,
|
||||||
CommandsModule,
|
CommandsModule,
|
||||||
|
AgentPresenceModule,
|
||||||
],
|
],
|
||||||
controllers: [HealthController, MetricsController],
|
controllers: [HealthController, MetricsController],
|
||||||
providers: [
|
providers: [
|
||||||
|
|||||||
@@ -1,4 +1,4 @@
|
|||||||
import { Body, Controller, Get, Param, Post, Query, Req, UnauthorizedException } from '@nestjs/common';
|
import { BadRequestException, Body, Controller, Get, Param, Patch, Post, Query, Req, UnauthorizedException } from '@nestjs/common';
|
||||||
import { ChannelsService } from './channels.service.js';
|
import { ChannelsService } from './channels.service.js';
|
||||||
|
|
||||||
// ApiKeyGuard attaches the introspected Center user id onto the request.
|
// ApiKeyGuard attaches the introspected Center user id onto the request.
|
||||||
@@ -32,11 +32,33 @@ export class ChannelsController {
|
|||||||
bypassUserIds: Array.isArray(body.bypassUserIds)
|
bypassUserIds: Array.isArray(body.bypassUserIds)
|
||||||
? (body.bypassUserIds as string[])
|
? (body.bypassUserIds as string[])
|
||||||
: [],
|
: [],
|
||||||
|
purpose: body.purpose as string | undefined,
|
||||||
},
|
},
|
||||||
userId,
|
userId,
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Patch a channel's free-form purpose. Body: { purpose: string }. Pass
|
||||||
|
// empty string to clear. Auth: channel member (or anyone for public
|
||||||
|
// channels, mirroring close()). Frontend doesn't call this today —
|
||||||
|
// intended for agent-side use (fabric-channel-set-purpose tool).
|
||||||
|
@Patch(':id')
|
||||||
|
patch(
|
||||||
|
@Req() req: AuthedRequest,
|
||||||
|
@Param('id') channelId: string,
|
||||||
|
@Body() body: Record<string, unknown>,
|
||||||
|
) {
|
||||||
|
const userId = req.userId ?? '';
|
||||||
|
if (!userId) throw new UnauthorizedException('missing user');
|
||||||
|
// Only `purpose` is patchable today. Future patchable fields would
|
||||||
|
// get their own typed branch; we explicitly NOT allow {} no-op patches
|
||||||
|
// because that signals a caller bug.
|
||||||
|
if (typeof body.purpose !== 'string') {
|
||||||
|
throw new BadRequestException('purpose (string) is required');
|
||||||
|
}
|
||||||
|
return this.channelsService.updatePurpose(channelId, userId, body.purpose);
|
||||||
|
}
|
||||||
|
|
||||||
// Move an order member into the bypass list (discuss/work only).
|
// Move an order member into the bypass list (discuss/work only).
|
||||||
@Post(':id/bypass')
|
@Post(':id/bypass')
|
||||||
bypass(
|
bypass(
|
||||||
|
|||||||
@@ -5,8 +5,9 @@ import { Channel } from '../entities/channel.entity.js';
|
|||||||
import { ChannelMember } from '../entities/channel-member.entity.js';
|
import { ChannelMember } from '../entities/channel-member.entity.js';
|
||||||
import { WakeMapping } from '../entities/wake-mapping.entity.js';
|
import { WakeMapping } from '../entities/wake-mapping.entity.js';
|
||||||
import { TurnService } from './turn.service.js';
|
import { TurnService } from './turn.service.js';
|
||||||
|
import { RealtimeGateway } from '../realtime/realtime.gateway.js';
|
||||||
|
|
||||||
const X_TYPES = ['general', 'work', 'report', 'discuss', 'triage', 'custom', 'dm'] as const;
|
const X_TYPES = ['general', 'work', 'report', 'discuss', 'triage', 'custom', 'dm', 'announce'] as const;
|
||||||
type XType = (typeof X_TYPES)[number];
|
type XType = (typeof X_TYPES)[number];
|
||||||
|
|
||||||
type CreateChannelInput = {
|
type CreateChannelInput = {
|
||||||
@@ -23,6 +24,10 @@ type CreateChannelInput = {
|
|||||||
// discuss/work only: members excluded from rotation (no wakeup unless
|
// discuss/work only: members excluded from rotation (no wakeup unless
|
||||||
// @-mentioned). order and bypass partition the members disjointly.
|
// @-mentioned). order and bypass partition the members disjointly.
|
||||||
bypassUserIds?: string[];
|
bypassUserIds?: string[];
|
||||||
|
// Free-form description of what this channel is for. Optional; agents
|
||||||
|
// typically fill it when creating, members can later edit via
|
||||||
|
// PATCH /api/channels/:id.
|
||||||
|
purpose?: string;
|
||||||
};
|
};
|
||||||
|
|
||||||
@Injectable()
|
@Injectable()
|
||||||
@@ -35,8 +40,34 @@ export class ChannelsService {
|
|||||||
@InjectRepository(WakeMapping)
|
@InjectRepository(WakeMapping)
|
||||||
private readonly wakeRepo: Repository<WakeMapping>,
|
private readonly wakeRepo: Repository<WakeMapping>,
|
||||||
private readonly turnService: TurnService,
|
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:
|
// Channels visible to a user within a guild:
|
||||||
// - every public channel of the guild (incl. ones created before the user
|
// - every public channel of the guild (incl. ones created before the user
|
||||||
// joined the guild), OR
|
// joined the guild), OR
|
||||||
@@ -93,6 +124,7 @@ export class ChannelsService {
|
|||||||
if (channel.xType === 'discuss' || channel.xType === 'work') {
|
if (channel.xType === 'discuss' || channel.xType === 'work') {
|
||||||
await this.turnService.onMemberAdded(channelId, userId);
|
await this.turnService.onMemberAdded(channelId, userId);
|
||||||
}
|
}
|
||||||
|
this.notifyMembership('joined', channelId, [userId], { xType: channel.xType });
|
||||||
}
|
}
|
||||||
return { status: 'ok', channelId, userId, member: true };
|
return { status: 'ok', channelId, userId, member: true };
|
||||||
}
|
}
|
||||||
@@ -102,11 +134,14 @@ export class ChannelsService {
|
|||||||
if (!channel) throw new NotFoundException('channel not found');
|
if (!channel) throw new NotFoundException('channel not found');
|
||||||
|
|
||||||
// remove every channel-scoped row that references this user
|
// 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 });
|
await this.wakeRepo.delete({ channelId, userId });
|
||||||
if (channel.xType === 'discuss' || channel.xType === 'work') {
|
if (channel.xType === 'discuss' || channel.xType === 'work') {
|
||||||
await this.turnService.onMemberRemoved(channelId, userId);
|
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 };
|
return { status: 'ok', channelId, userId, member: false };
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -135,6 +170,9 @@ export class ChannelsService {
|
|||||||
// allowed (create() always makes a fresh one, no dedup).
|
// allowed (create() always makes a fresh one, no dedup).
|
||||||
const isPublic = xType === 'dm' ? false : Boolean(input.isPublic);
|
const isPublic = xType === 'dm' ? false : Boolean(input.isPublic);
|
||||||
|
|
||||||
|
const purposeRaw = String(input.purpose ?? '').trim();
|
||||||
|
const purpose = purposeRaw === '' ? null : purposeRaw;
|
||||||
|
|
||||||
const channel = await this.channelRepo.save(
|
const channel = await this.channelRepo.save(
|
||||||
this.channelRepo.create({
|
this.channelRepo.create({
|
||||||
guildId,
|
guildId,
|
||||||
@@ -143,6 +181,7 @@ export class ChannelsService {
|
|||||||
kind: input.kind === 'announcement' ? 'announcement' : 'text',
|
kind: input.kind === 'announcement' ? 'announcement' : 'text',
|
||||||
isPrivate: !isPublic,
|
isPrivate: !isPublic,
|
||||||
isPublic,
|
isPublic,
|
||||||
|
purpose,
|
||||||
lastSeq: 0,
|
lastSeq: 0,
|
||||||
}),
|
}),
|
||||||
);
|
);
|
||||||
@@ -160,6 +199,12 @@ export class ChannelsService {
|
|||||||
[...memberIds].map((userId) => this.memberRepo.create({ channelId: channel.id, userId })),
|
[...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
|
// wake_mapping: triage -> the on-duty user; custom -> each listener
|
||||||
const wakeUserIds = new Set<string>();
|
const wakeUserIds = new Set<string>();
|
||||||
if (xType === 'triage') wakeUserIds.add(onDuty);
|
if (xType === 'triage') wakeUserIds.add(onDuty);
|
||||||
@@ -182,6 +227,27 @@ export class ChannelsService {
|
|||||||
return channel;
|
return channel;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Update a channel's free-form purpose. Any channel member may do this
|
||||||
|
// (or any guild user if the channel is public, mirroring closeChannel's
|
||||||
|
// member-or-public rule). Pass an empty string to clear.
|
||||||
|
async updatePurpose(channelId: string, actorUserId: string, purpose: string) {
|
||||||
|
const channel = await this.channelRepo.findOne({ where: { id: channelId } });
|
||||||
|
if (!channel) throw new NotFoundException('channel not found');
|
||||||
|
const member = await this.memberRepo.findOne({ where: { channelId, userId: actorUserId } });
|
||||||
|
if (!member && !channel.isPublic) {
|
||||||
|
throw new ForbiddenException('not a channel member');
|
||||||
|
}
|
||||||
|
const trimmed = String(purpose ?? '').trim();
|
||||||
|
channel.purpose = trimmed === '' ? null : trimmed;
|
||||||
|
const saved = await this.channelRepo.save(channel);
|
||||||
|
return {
|
||||||
|
id: saved.id,
|
||||||
|
name: saved.name,
|
||||||
|
xType: saved.xType,
|
||||||
|
purpose: saved.purpose,
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
// Move an order member into the bypass list (discuss/work only).
|
// Move an order member into the bypass list (discuss/work only).
|
||||||
// Any channel member may do this.
|
// Any channel member may do this.
|
||||||
async moveToBypass(channelId: string, actorUserId: string, targetUserId: string) {
|
async moveToBypass(channelId: string, actorUserId: string, targetUserId: string) {
|
||||||
|
|||||||
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);
|
||||||
|
});
|
||||||
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
|
// Resolve <@user.name:NAME> names to userIds within this guild node via
|
||||||
// Center. Unresolved names are simply absent from the returned map.
|
// Center. Unresolved names are simply absent from the returned map.
|
||||||
export async function resolveUserNames(names: string[]): Promise<Record<string, string>> {
|
export async function resolveUserNames(names: string[]): Promise<Record<string, string>> {
|
||||||
|
|||||||
@@ -14,6 +14,7 @@ import { IdempotencyRecord } from './entities/idempotency-record.entity.js';
|
|||||||
import { StoredFile } from './entities/stored-file.entity.js';
|
import { StoredFile } from './entities/stored-file.entity.js';
|
||||||
import { ChannelCanvas } from './entities/channel-canvas.entity.js';
|
import { ChannelCanvas } from './entities/channel-canvas.entity.js';
|
||||||
import { GuildCommand } from './entities/guild-command.entity.js';
|
import { GuildCommand } from './entities/guild-command.entity.js';
|
||||||
|
import { AgentPresence } from './entities/agent-presence.entity.js';
|
||||||
|
|
||||||
export const buildTypeOrmConfig = (): TypeOrmModuleOptions => ({
|
export const buildTypeOrmConfig = (): TypeOrmModuleOptions => ({
|
||||||
type: 'mysql',
|
type: 'mysql',
|
||||||
@@ -38,6 +39,7 @@ export const buildTypeOrmConfig = (): TypeOrmModuleOptions => ({
|
|||||||
StoredFile,
|
StoredFile,
|
||||||
ChannelCanvas,
|
ChannelCanvas,
|
||||||
GuildCommand,
|
GuildCommand,
|
||||||
|
AgentPresence,
|
||||||
],
|
],
|
||||||
synchronize: (process.env.FABRIC_BACKEND_GUILD_DB_SYNC ?? 'true') === 'true',
|
synchronize: (process.env.FABRIC_BACKEND_GUILD_DB_SYNC ?? 'true') === 'true',
|
||||||
logging: (process.env.FABRIC_BACKEND_GUILD_DB_LOGGING ?? 'false') === 'true',
|
logging: (process.env.FABRIC_BACKEND_GUILD_DB_LOGGING ?? 'false') === 'true',
|
||||||
|
|||||||
35
src/entities/agent-presence.entity.ts
Normal file
35
src/entities/agent-presence.entity.ts
Normal file
@@ -0,0 +1,35 @@
|
|||||||
|
import { Column, Entity, PrimaryColumn, UpdateDateColumn } from 'typeorm';
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Per-user (typically agent) presence cache.
|
||||||
|
*
|
||||||
|
* Populated by Fabric.OpenclawPlugin's presence-sync loop: every ~30s
|
||||||
|
* it reads each connected agent's HF status from the cross-plugin
|
||||||
|
* `globalThis.__hfAgentStatus.get(agentId)` (exposed by
|
||||||
|
* HarborForge.OpenclawPlugin) and pushes diffs via
|
||||||
|
* `PUT /agents/:userId/presence`.
|
||||||
|
*
|
||||||
|
* Used by `RealtimeGateway.computeDelivery` for `announce`-type
|
||||||
|
* channels to skip delivery to recipients whose status is `busy`.
|
||||||
|
* Defaults to `unknown` if no row exists (treated as not-busy).
|
||||||
|
*/
|
||||||
|
@Entity('agent_presences')
|
||||||
|
export class AgentPresence {
|
||||||
|
// Same id as the Fabric Center user id (UUID v4 string, char(36)).
|
||||||
|
@PrimaryColumn({ type: 'char', length: 36 })
|
||||||
|
userId!: string;
|
||||||
|
|
||||||
|
@Column({
|
||||||
|
type: 'enum',
|
||||||
|
enum: ['idle', 'on_call', 'busy', 'exhausted', 'offline', 'unknown'],
|
||||||
|
default: 'unknown',
|
||||||
|
})
|
||||||
|
status!: 'idle' | 'on_call' | 'busy' | 'exhausted' | 'offline' | 'unknown';
|
||||||
|
|
||||||
|
/** Free-text source tag for debugging ("hf-plugin", "manual", etc.). */
|
||||||
|
@Column({ type: 'varchar', length: 64, default: 'unknown' })
|
||||||
|
source!: string;
|
||||||
|
|
||||||
|
@UpdateDateColumn()
|
||||||
|
updatedAt!: Date;
|
||||||
|
}
|
||||||
@@ -16,13 +16,23 @@ export class Channel {
|
|||||||
@Column({
|
@Column({
|
||||||
name: 'x_type',
|
name: 'x_type',
|
||||||
type: 'enum',
|
type: 'enum',
|
||||||
enum: ['general', 'work', 'report', 'discuss', 'triage', 'custom', 'dm'],
|
enum: ['general', 'work', 'report', 'discuss', 'triage', 'custom', 'dm', 'announce'],
|
||||||
})
|
})
|
||||||
xType!: 'general' | 'work' | 'report' | 'discuss' | 'triage' | 'custom' | 'dm';
|
xType!: 'general' | 'work' | 'report' | 'discuss' | 'triage' | 'custom' | 'dm' | 'announce';
|
||||||
|
|
||||||
@Column({ type: 'varchar', length: 16, default: 'text' })
|
@Column({ type: 'varchar', length: 16, default: 'text' })
|
||||||
kind!: 'text' | 'announcement';
|
kind!: 'text' | 'announcement';
|
||||||
|
|
||||||
|
// Free-form description of what this channel is for — what topics get
|
||||||
|
// posted, who participates, why it exists. Surfaced via GET /api/channels
|
||||||
|
// so agents can pick a channel by intent ("which announce channel is for
|
||||||
|
// debate broadcasts?") without channel id hard-coded into workflows.
|
||||||
|
// Any channel member can set it via PATCH /api/channels/:id (writes
|
||||||
|
// require membership the same way moveToBypass / close do). The frontend
|
||||||
|
// create form does NOT post this today — purpose stays optional.
|
||||||
|
@Column({ type: 'text', nullable: true })
|
||||||
|
purpose!: string | null;
|
||||||
|
|
||||||
@Column({ type: 'boolean', default: false })
|
@Column({ type: 'boolean', default: false })
|
||||||
isPrivate!: boolean;
|
isPrivate!: boolean;
|
||||||
|
|
||||||
|
|||||||
@@ -21,6 +21,7 @@ import { ChannelMember } from '../entities/channel-member.entity.js';
|
|||||||
import { Message } from '../entities/message.entity.js';
|
import { Message } from '../entities/message.entity.js';
|
||||||
import { IdempotencyRecord } from '../entities/idempotency-record.entity.js';
|
import { IdempotencyRecord } from '../entities/idempotency-record.entity.js';
|
||||||
import { WakeMapping } from '../entities/wake-mapping.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 { parseSlashCommand } from '../channels/slash-commands.js';
|
||||||
import { parseMentions, extractNameMentions, replaceNameMentions } from '../channels/mentions.js';
|
import { parseMentions, extractNameMentions, replaceNameMentions } from '../channels/mentions.js';
|
||||||
import { resolveUserNames } from '../common/center-auth.js';
|
import { resolveUserNames } from '../common/center-auth.js';
|
||||||
@@ -50,6 +51,7 @@ export class MessagingController {
|
|||||||
private readonly turn: TurnService,
|
private readonly turn: TurnService,
|
||||||
private readonly events: EventsService,
|
private readonly events: EventsService,
|
||||||
private readonly realtime: RealtimeGateway,
|
private readonly realtime: RealtimeGateway,
|
||||||
|
private readonly adminCache: AdminCacheService,
|
||||||
) {}
|
) {}
|
||||||
|
|
||||||
private async getIdempotentResponse(
|
private async getIdempotentResponse(
|
||||||
@@ -164,6 +166,11 @@ export class MessagingController {
|
|||||||
// Guild C-1: caller must be a participant of the channel, and the
|
// Guild C-1: caller must be a participant of the channel, and the
|
||||||
// author is always the authenticated user — body.authorUserId is
|
// author is always the authenticated user — body.authorUserId is
|
||||||
// ignored so a caller can never post as someone else.
|
// ignored so a caller can never post as someone else.
|
||||||
|
//
|
||||||
|
// announce channels: any participant can POST. Use case is one-off
|
||||||
|
// recruitment / broadcast messages posted by the agent that just
|
||||||
|
// created the originating topic (e.g. dialectic invites). No
|
||||||
|
// server-side privileged path — author is always a real user.
|
||||||
const userId = String(req.userId ?? '');
|
const userId = String(req.userId ?? '');
|
||||||
if (!userId) throw new ForbiddenException('missing user');
|
if (!userId) throw new ForbiddenException('missing user');
|
||||||
const channel = await this.assertParticipant(channelId, userId);
|
const channel = await this.assertParticipant(channelId, userId);
|
||||||
@@ -225,16 +232,19 @@ export class MessagingController {
|
|||||||
const decision = await this.turn.onNormalMessage(channelId, authorUserId, mentionIds);
|
const decision = await this.turn.onNormalMessage(channelId, authorUserId, mentionIds);
|
||||||
await this.realtime.emitMessageTargeted(channelId, responseBody, decision.wakeupUserId);
|
await this.realtime.emitMessageTargeted(channelId, responseBody, decision.wakeupUserId);
|
||||||
} else {
|
} else {
|
||||||
// general/report/triage/custom: wakeup from x_type + wake_mapping;
|
// general/report/triage/custom: 3-state delivery
|
||||||
// general also honors the message's at-list
|
// (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 wakeRows = await this.wakeRepo.find({ where: { channelId } });
|
||||||
const wakeUserIds = new Set(wakeRows.map((w) => w.userId));
|
const wakeUserIds = new Set(wakeRows.map((w) => w.userId));
|
||||||
const mentionUserIds = new Set(mentionIds.filter((id) => id !== authorUserId));
|
const mentionUserIds = new Set(mentionIds.filter((id) => id !== authorUserId));
|
||||||
|
const admin = await this.adminCache.get();
|
||||||
await this.realtime.emitMessageCreated(channelId, responseBody, {
|
await this.realtime.emitMessageCreated(channelId, responseBody, {
|
||||||
xType,
|
xType,
|
||||||
authorUserId,
|
authorUserId,
|
||||||
wakeUserIds,
|
wakeUserIds,
|
||||||
mentionUserIds,
|
mentionUserIds,
|
||||||
|
adminUserId: admin?.userId ?? null,
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -6,9 +6,12 @@ import { ChannelMember } from '../entities/channel-member.entity.js';
|
|||||||
import { Message } from '../entities/message.entity.js';
|
import { Message } from '../entities/message.entity.js';
|
||||||
import { IdempotencyRecord } from '../entities/idempotency-record.entity.js';
|
import { IdempotencyRecord } from '../entities/idempotency-record.entity.js';
|
||||||
import { WakeMapping } from '../entities/wake-mapping.entity.js';
|
import { WakeMapping } from '../entities/wake-mapping.entity.js';
|
||||||
|
import { AdminCacheService } from '../common/admin-cache.service.js';
|
||||||
|
|
||||||
@Module({
|
@Module({
|
||||||
imports: [TypeOrmModule.forFeature([Channel, ChannelMember, Message, IdempotencyRecord, WakeMapping])],
|
imports: [TypeOrmModule.forFeature([Channel, ChannelMember, Message, IdempotencyRecord, WakeMapping])],
|
||||||
controllers: [MessagingController],
|
controllers: [MessagingController],
|
||||||
|
providers: [AdminCacheService],
|
||||||
|
exports: [AdminCacheService],
|
||||||
})
|
})
|
||||||
export class MessagingModule {}
|
export class MessagingModule {}
|
||||||
|
|||||||
@@ -11,17 +11,94 @@ import { Logger } from '@nestjs/common';
|
|||||||
import { Server, Socket } from 'socket.io';
|
import { Server, Socket } from 'socket.io';
|
||||||
import { introspectGuildToken } from '../common/center-auth.js';
|
import { introspectGuildToken } from '../common/center-auth.js';
|
||||||
|
|
||||||
type XType = 'general' | 'work' | 'report' | 'discuss' | 'triage' | 'custom' | 'dm';
|
type XType = 'general' | 'work' | 'report' | 'discuss' | 'triage' | 'custom' | 'dm' | 'announce';
|
||||||
|
|
||||||
// Wakeup for non-rotating channels only (general/report/triage/custom).
|
/**
|
||||||
// discuss/work go through TurnService + emitMessageTargeted, never here.
|
* Cross-presence info needed by `announce`-type delivery: a recipient
|
||||||
// Precedence:
|
* with hf-side status === 'busy' has the message discarded silently
|
||||||
// 1. the author never gets woken by their own message
|
* (don't enter their session, no UI emit). Other statuses + non-announce
|
||||||
// 2. triage/custom: only wake users in the channel's wake_mapping
|
* channels are unaffected. Presence is sourced from the
|
||||||
// (mentions change nothing here)
|
* `agent_presences` table populated by Fabric.OpenclawPlugin's
|
||||||
// 3. general: if the message has an at-list, wake only the at'd users;
|
* presence-sync loop (which reads from HF plugin's `__hfAgentStatus`).
|
||||||
// otherwise wake everyone
|
*/
|
||||||
// 4. report (and anything else): wake nobody
|
export type PresenceStatus = 'idle' | 'on_call' | 'busy' | 'exhausted' | 'offline' | 'unknown';
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 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;
|
||||||
|
/** Recipient's current presence; only consulted for `announce` xType. Defaults to 'unknown' (treated as not-busy). */
|
||||||
|
recipientPresence?: PresenceStatus;
|
||||||
|
}
|
||||||
|
|
||||||
|
export function computeDelivery(args: ComputeDeliveryArgs): DeliveryDecision {
|
||||||
|
const { xType, recipientUserId, authorUserId, wakeUserIds, mentionUserIds, adminUserId, recipientPresence } = 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';
|
||||||
|
case 'announce':
|
||||||
|
// System-broadcast channels (e.g. Dialectic topic announcements).
|
||||||
|
// Recipients with HF status === 'busy' have the message discarded
|
||||||
|
// silently — busy agents should not be distracted by signup pings
|
||||||
|
// they can't act on. All other presences (idle/on_call/exhausted/
|
||||||
|
// offline/unknown) get the message as 'observer' (no wake): the
|
||||||
|
// channel itself is browsable; agents proactively decide what to
|
||||||
|
// do with announcements when they next look at their inbox.
|
||||||
|
if (recipientPresence === 'busy') return 'skip';
|
||||||
|
return 'observer';
|
||||||
|
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: {
|
export function computeWakeup(args: {
|
||||||
xType: XType;
|
xType: XType;
|
||||||
recipientUserId: string;
|
recipientUserId: string;
|
||||||
@@ -29,23 +106,7 @@ export function computeWakeup(args: {
|
|||||||
wakeUserIds: Set<string>;
|
wakeUserIds: Set<string>;
|
||||||
mentionUserIds?: Set<string>;
|
mentionUserIds?: Set<string>;
|
||||||
}): boolean {
|
}): boolean {
|
||||||
const { xType, recipientUserId, authorUserId, wakeUserIds, mentionUserIds } = args;
|
return computeDelivery(args) === 'wake';
|
||||||
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);
|
|
||||||
case 'dm':
|
|
||||||
// 1:1 conversation: every non-author participant is always woken.
|
|
||||||
return true;
|
|
||||||
default:
|
|
||||||
return false;
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
@WebSocketGateway({
|
@WebSocketGateway({
|
||||||
@@ -61,6 +122,12 @@ export class RealtimeGateway implements OnGatewayConnection, OnGatewayDisconnect
|
|||||||
private readonly logger = new Logger(RealtimeGateway.name);
|
private readonly logger = new Logger(RealtimeGateway.name);
|
||||||
private readonly onlineUsers = new Set<string>();
|
private readonly onlineUsers = new Set<string>();
|
||||||
|
|
||||||
|
// Optional: injected at module wiring time. Used by emitMessageCreated
|
||||||
|
// to pre-load recipient presence for announce-type channels.
|
||||||
|
// Typed loosely to avoid a circular import between realtime and agents
|
||||||
|
// modules; the actual interface lives in agents/agent-presence.service.
|
||||||
|
presence?: { getStatusMap(ids: string[]): Promise<Map<string, PresenceStatus>> };
|
||||||
|
|
||||||
private userIdFromClient(client: Socket): string {
|
private userIdFromClient(client: Socket): string {
|
||||||
const authUser = client.handshake.auth?.userId;
|
const authUser = client.handshake.auth?.userId;
|
||||||
const headerUser = client.handshake.headers['x-user-id'];
|
const headerUser = client.handshake.headers['x-user-id'];
|
||||||
@@ -96,6 +163,10 @@ export class RealtimeGateway implements OnGatewayConnection, OnGatewayDisconnect
|
|||||||
const userId = result.user.id || this.userIdFromClient(client);
|
const userId = result.user.id || this.userIdFromClient(client);
|
||||||
client.data.userId = userId;
|
client.data.userId = userId;
|
||||||
this.onlineUsers.add(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', {
|
this.server.emit('presence.online', {
|
||||||
userId,
|
userId,
|
||||||
onlineCount: this.onlineUsers.size,
|
onlineCount: this.onlineUsers.size,
|
||||||
@@ -171,7 +242,19 @@ export class RealtimeGateway implements OnGatewayConnection, OnGatewayDisconnect
|
|||||||
this.server.to(`channel:${channelId}`).emit(event, data);
|
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, and by announce
|
||||||
|
// channels to suppress delivery to recipients whose presence is busy.
|
||||||
async emitMessageCreated(
|
async emitMessageCreated(
|
||||||
channelId: string,
|
channelId: string,
|
||||||
data: Record<string, unknown>,
|
data: Record<string, unknown>,
|
||||||
@@ -180,19 +263,41 @@ export class RealtimeGateway implements OnGatewayConnection, OnGatewayDisconnect
|
|||||||
authorUserId: string;
|
authorUserId: string;
|
||||||
wakeUserIds: Set<string>;
|
wakeUserIds: Set<string>;
|
||||||
mentionUserIds?: Set<string>;
|
mentionUserIds?: Set<string>;
|
||||||
|
/** Single Center-scoped admin userId (or null). */
|
||||||
|
adminUserId?: string | null;
|
||||||
},
|
},
|
||||||
): Promise<void> {
|
): Promise<void> {
|
||||||
const sockets = await this.server.in(`channel:${channelId}`).fetchSockets();
|
const sockets = await this.server.in(`channel:${channelId}`).fetchSockets();
|
||||||
|
|
||||||
|
// For announce-type channels, pre-load presence for all recipients
|
||||||
|
// in one query so the per-recipient loop doesn't fan out to N round
|
||||||
|
// trips. For other xTypes, presence is irrelevant — skip the lookup.
|
||||||
|
let presenceMap: Map<string, PresenceStatus> | undefined;
|
||||||
|
if (ctx.xType === 'announce' && this.presence) {
|
||||||
|
const recipientIds = sockets
|
||||||
|
.map((s) => (typeof s.data.userId === 'string' ? (s.data.userId as string) : ''))
|
||||||
|
.filter((id) => id && !id.startsWith('anon:'));
|
||||||
|
presenceMap = await this.presence.getStatusMap(recipientIds);
|
||||||
|
}
|
||||||
|
|
||||||
for (const s of sockets) {
|
for (const s of sockets) {
|
||||||
const recipientUserId = typeof s.data.userId === 'string' ? s.data.userId : `anon:${s.id}`;
|
const recipientUserId = typeof s.data.userId === 'string' ? s.data.userId : `anon:${s.id}`;
|
||||||
const wakeup = computeWakeup({
|
const decision = computeDelivery({
|
||||||
xType: ctx.xType,
|
xType: ctx.xType,
|
||||||
recipientUserId,
|
recipientUserId,
|
||||||
authorUserId: ctx.authorUserId,
|
authorUserId: ctx.authorUserId,
|
||||||
wakeUserIds: ctx.wakeUserIds,
|
wakeUserIds: ctx.wakeUserIds,
|
||||||
mentionUserIds: ctx.mentionUserIds,
|
mentionUserIds: ctx.mentionUserIds,
|
||||||
|
adminUserId: ctx.adminUserId,
|
||||||
|
recipientPresence: presenceMap?.get(recipientUserId) ?? 'unknown',
|
||||||
|
});
|
||||||
|
if (decision === 'skip') continue;
|
||||||
|
s.emit('message.created', {
|
||||||
|
...data,
|
||||||
|
channelId,
|
||||||
|
wakeup: decision === 'wake',
|
||||||
|
xType: ctx.xType,
|
||||||
});
|
});
|
||||||
s.emit('message.created', { ...data, channelId, wakeup, xType: ctx.xType });
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -1,9 +1,25 @@
|
|||||||
import { Global, Module } from '@nestjs/common';
|
import { Global, Module, OnModuleInit } from '@nestjs/common';
|
||||||
import { RealtimeGateway } from './realtime.gateway.js';
|
import { RealtimeGateway } from './realtime.gateway.js';
|
||||||
|
import { AgentPresenceModule } from '../agents/agent-presence.module.js';
|
||||||
|
import { AgentPresenceService } from '../agents/agent-presence.service.js';
|
||||||
|
|
||||||
@Global()
|
@Global()
|
||||||
@Module({
|
@Module({
|
||||||
|
imports: [AgentPresenceModule],
|
||||||
providers: [RealtimeGateway],
|
providers: [RealtimeGateway],
|
||||||
exports: [RealtimeGateway],
|
exports: [RealtimeGateway],
|
||||||
})
|
})
|
||||||
export class RealtimeModule {}
|
export class RealtimeModule implements OnModuleInit {
|
||||||
|
// Wire presence into the gateway at startup. Using assignment (vs
|
||||||
|
// constructor injection) keeps the gateway free of the agents-module
|
||||||
|
// import — no risk of circular dependency, and announce-channel
|
||||||
|
// delivery degrades gracefully (presence stays undefined → 'unknown'
|
||||||
|
// status → no busy-discard) if AgentPresenceModule is ever removed.
|
||||||
|
constructor(
|
||||||
|
private readonly gateway: RealtimeGateway,
|
||||||
|
private readonly presence: AgentPresenceService,
|
||||||
|
) {}
|
||||||
|
onModuleInit(): void {
|
||||||
|
this.gateway.presence = this.presence;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user