feat(guild): wake_mapping, per-recipient wakeup, discuss/work turn engine, channel join/leave
- wake_mapping table; triage onDuty (auto-added member) / custom listeners - per-recipient wakeup metadata on message.created (one message-id; added only at push). Rules: author=false; triage/custom=wake_mapping only; general=all; report=none - discuss/work rotation: channel_turn_state (order/currentSpeaker/round events/cross-round no-reply streak); null activation, queue-jump, /no-reply pass, all-/no-reply pause, end-of-round shuffle (trailing no-reply run to tail, head shuffled, first != last normal speaker) - slash-command registry (/no-reply, /force-proceed); registered commands intercepted and never delivered; guild-authored /ack persisted - POST /channels/:id/join|leave; leave cleans channel_members, wake_mapping and turn-state order Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -16,6 +16,9 @@ import { CreateMessageDto } from './dto.create-message.dto';
|
||||
import { Channel } from '../entities/channel.entity';
|
||||
import { Message } from '../entities/message.entity';
|
||||
import { IdempotencyRecord } from '../entities/idempotency-record.entity';
|
||||
import { WakeMapping } from '../entities/wake-mapping.entity';
|
||||
import { parseSlashCommand } from '../channels/slash-commands';
|
||||
import { TurnService } from '../channels/turn.service';
|
||||
import { EventsService } from '../events/events.service';
|
||||
import { clampLimit, computeNextExpectedSeq } from './pagination.util';
|
||||
import { RealtimeGateway } from '../realtime/realtime.gateway';
|
||||
@@ -34,6 +37,9 @@ export class MessagingController {
|
||||
private readonly messageRepo: Repository<Message>,
|
||||
@InjectRepository(IdempotencyRecord)
|
||||
private readonly idemRepo: Repository<IdempotencyRecord>,
|
||||
@InjectRepository(WakeMapping)
|
||||
private readonly wakeRepo: Repository<WakeMapping>,
|
||||
private readonly turn: TurnService,
|
||||
private readonly events: EventsService,
|
||||
private readonly realtime: RealtimeGateway,
|
||||
) {}
|
||||
@@ -77,6 +83,52 @@ export class MessagingController {
|
||||
};
|
||||
}
|
||||
|
||||
// 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(
|
||||
channelId: string,
|
||||
input: { authorUserId: string; content: string; clientMessageId?: string | null; replyToMessageId?: string | null; mentions?: string[]; attachments?: Array<{ url: string; name?: string; mimeType?: string }> },
|
||||
): Promise<Message> {
|
||||
return this.dataSource.transaction(async (manager) => {
|
||||
const channel = await manager.findOne(Channel, {
|
||||
where: { id: channelId },
|
||||
lock: { mode: 'pessimistic_write' },
|
||||
});
|
||||
if (!channel) {
|
||||
throw new NotFoundException('channel not found');
|
||||
}
|
||||
const nextSeq = channel.lastSeq + 1;
|
||||
channel.lastSeq = nextSeq;
|
||||
await manager.save(Channel, channel);
|
||||
|
||||
const messageId = input.clientMessageId ?? `m-${channelId}-${nextSeq}`;
|
||||
const row = manager.create(Message, {
|
||||
messageId,
|
||||
channelId,
|
||||
conversationId: null,
|
||||
authorUserId: input.authorUserId,
|
||||
seq: nextSeq,
|
||||
content: input.content,
|
||||
replyToMessageId: input.replyToMessageId ?? null,
|
||||
mentions: input.mentions ?? [],
|
||||
attachments: input.attachments ?? [],
|
||||
editedAt: null,
|
||||
deletedAt: null,
|
||||
isDeleted: false,
|
||||
});
|
||||
return manager.save(Message, row);
|
||||
});
|
||||
}
|
||||
|
||||
// Emits a guild-authored /ack message to the channel; wakeup=true only for
|
||||
// the new current speaker (null = nobody). One message-id; persisted.
|
||||
private async emitAck(channelId: string, wakeupUserId: string | null): Promise<void> {
|
||||
const ack = await this.persistMessage(channelId, { authorUserId: 'guild', content: '/ack' });
|
||||
const body = this.toView(ack) as Record<string, unknown>;
|
||||
await this.events.emit({ eventType: 'message.created', channelId, actorId: 'guild', data: body });
|
||||
await this.realtime.emitMessageTargeted(channelId, body, wakeupUserId);
|
||||
}
|
||||
|
||||
@Post()
|
||||
async create(
|
||||
@Param('id') channelId: string,
|
||||
@@ -87,35 +139,34 @@ export class MessagingController {
|
||||
const existed = await this.getIdempotentResponse(scope, idempotencyKey);
|
||||
if (existed) return existed;
|
||||
|
||||
const message = await this.dataSource.transaction(async (manager) => {
|
||||
const channel = await manager.findOne(Channel, {
|
||||
where: { id: channelId },
|
||||
lock: { mode: 'pessimistic_write' },
|
||||
});
|
||||
if (!channel) {
|
||||
throw new NotFoundException('channel not found');
|
||||
const channel = await this.channelRepo.findOne({ where: { id: channelId } });
|
||||
if (!channel) throw new NotFoundException('channel not found');
|
||||
const xType = channel.xType ?? 'general';
|
||||
const isRotating = xType === 'discuss' || xType === 'work';
|
||||
const authorUserId = String(body.authorUserId ?? 'anonymous');
|
||||
|
||||
// ---- command interception: registered slash commands are never delivered
|
||||
const cmd = parseSlashCommand(body.content ?? '');
|
||||
if (cmd) {
|
||||
if (isRotating && cmd.name === 'no-reply') {
|
||||
const { ack } = await this.turn.onNoReply(channelId, authorUserId);
|
||||
if (ack) await this.emitAck(channelId, ack.wakeupUserId);
|
||||
} else if (isRotating && cmd.name === 'force-proceed') {
|
||||
const { ack } = await this.turn.onForceProceed(channelId);
|
||||
if (ack) await this.emitAck(channelId, ack.wakeupUserId);
|
||||
}
|
||||
// non-rotating channels (or no effect): swallowed, nothing delivered
|
||||
return { status: 'command', command: cmd.name };
|
||||
}
|
||||
|
||||
const nextSeq = channel.lastSeq + 1;
|
||||
channel.lastSeq = nextSeq;
|
||||
await manager.save(Channel, channel);
|
||||
|
||||
const messageId = body.clientMessageId ?? `m-${channelId}-${nextSeq}`;
|
||||
const row = manager.create(Message, {
|
||||
messageId,
|
||||
channelId,
|
||||
conversationId: null,
|
||||
authorUserId: body.authorUserId ?? 'anonymous',
|
||||
seq: nextSeq,
|
||||
content: body.content,
|
||||
replyToMessageId: body.replyToMessageId ?? null,
|
||||
mentions: body.mentions ?? [],
|
||||
attachments: body.attachments ?? [],
|
||||
editedAt: null,
|
||||
deletedAt: null,
|
||||
isDeleted: false,
|
||||
});
|
||||
return manager.save(Message, row);
|
||||
// ---- normal message
|
||||
const message = await this.persistMessage(channelId, {
|
||||
authorUserId,
|
||||
content: body.content,
|
||||
clientMessageId: body.clientMessageId,
|
||||
replyToMessageId: body.replyToMessageId,
|
||||
mentions: body.mentions,
|
||||
attachments: body.attachments,
|
||||
});
|
||||
|
||||
const responseBody = this.toView(message) as Record<string, unknown>;
|
||||
@@ -124,10 +175,24 @@ export class MessagingController {
|
||||
await this.events.emit({
|
||||
eventType: 'message.created',
|
||||
channelId,
|
||||
actorId: body.authorUserId ?? 'anonymous',
|
||||
actorId: authorUserId,
|
||||
data: responseBody,
|
||||
});
|
||||
this.realtime.emitChannelEvent(channelId, 'message.created', responseBody);
|
||||
|
||||
if (isRotating) {
|
||||
// discuss/work: rotation decides the single wakeup target
|
||||
const decision = await this.turn.onNormalMessage(channelId, authorUserId);
|
||||
await this.realtime.emitMessageTargeted(channelId, responseBody, decision.wakeupUserId);
|
||||
} else {
|
||||
// general/report/triage/custom: wakeup from x_type + wake_mapping
|
||||
const wakeRows = await this.wakeRepo.find({ where: { channelId } });
|
||||
const wakeUserIds = new Set(wakeRows.map((w) => w.userId));
|
||||
await this.realtime.emitMessageCreated(channelId, responseBody, {
|
||||
xType,
|
||||
authorUserId,
|
||||
wakeUserIds,
|
||||
});
|
||||
}
|
||||
|
||||
return responseBody;
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user