import { BadRequestException, ConflictException, GoneException, Inject, Injectable, InternalServerErrorException, NotFoundException } from "@nestjs/common"; import { DEFAULT_MANUAL_LANGUAGE, MANUAL_LANGUAGES, isManualLanguage, type ManualLanguage, } from "@tactical-media-broadcast-hub/core"; import type { PoolClient } from "pg"; import { appendTenantAudit } from "../audit/audit.service.js"; import { withSystem, withTenant } from "../db/client.js"; /** * Collectives — P1 of docs/design/2026-10-12-collectives-architecture.md. * * A collective IS a group: `groups.parent_id` self-references (root groups are * collectives, children are subcollectives) and `groups.kind` discriminates them * from the 008-era sharing-scope groups ('plain'). The room set is mapped in * `group_rooms` (one open, one closed room per collective plus the founder's * onboarding room), provisioned through PROVISIONER_PORT so the real Matrix * implementation is swap-faked in tests. * * Provisioning order: rooms are created on the homeserver FIRST, then the group, * the founder's group_members row and the group_rooms mappings land in ONE * tenant transaction. A remote side effect cannot be rolled back, so the DB is * only written once the full room set exists — a failed homeserver leaves no * collective at all (orphan rooms are invisible; a collective with missing * rooms would be a lie). */ export const COLLECTIVE_ROOM_TYPES = ["open", "closed", "onboarding"] as const; export type CollectiveRoomType = (typeof COLLECTIVE_ROOM_TYPES)[number]; export type GroupKind = "collective" | "subcollective" | "plain"; /** Power levels the provisioner pins at creation (design §Provisioning). */ export const FOUNDER_POWER_LEVEL = 100; export const MEMBER_POWER_LEVEL = 0; export interface CollectiveActor { userId: string; role: string; } export interface ProvisionedRoom { roomId: string; } /** A whole provisioned room set, exactly as the provisioner produced it. */ export interface ProvisionedRoomSet { open: ProvisionedRoom; closed: ProvisionedRoom; onboarding: ProvisionedRoom; } export interface ProvisionCollectiveInput { workspaceId: string; /** Informational only at provision time; never used as a foreign key. */ groupId: string; name: string; language: ManualLanguage; founderUserId: string; /** Founder's Matrix user when this deployment knows it; null = skip invites. */ founderMxid: string | null; } /** * The seam between the Hub and the homeserver. MatrixProvisioner is the only * implementation that knows Matrix exists; the service, the controller and * every test work against this interface (see InMemoryProvisioner in tests). * * Throws ProvisionerUnavailableError when the homeserver cannot be reached or * refuses — the service maps that to 503 without persisting anything. */ export interface ProvisionerPort { readonly botUserId: string; provisionCollective(input: ProvisionCollectiveInput): Promise; } export class ProvisionerUnavailableError extends Error { constructor(message: string) { super(message); this.name = "ProvisionerUnavailableError"; } } /** DI token; the module binds MatrixProvisioner (or a fake) to it. */ export const PROVISIONER_PORT = Symbol("PROVISIONER_PORT"); export interface CollectiveRoomSummary { roomType: CollectiveRoomType; roomId: string; } export interface CollectiveNode { id: string; name: string; kind: GroupKind; language: ManualLanguage; parentId: string | null; description: string; createdAt: string; memberCount: number; rooms: CollectiveRoomSummary[]; children: CollectiveNode[]; } export interface CreateCollectiveInput { name: string; language: ManualLanguage; description?: string; } const UNIQUE_VIOLATION = "23505"; const FK_VIOLATION = "23503"; export function pgCode(err: unknown): string | undefined { if (typeof err !== "object" || err === null || !("code" in err)) return undefined; const code = (err as { code: unknown }).code; return typeof code === "string" ? code : undefined; } /** Language parsing shared by the controller and the service. */ export function parseLanguage(value: unknown): ManualLanguage { if (value === undefined || value === null || value === "") return DEFAULT_MANUAL_LANGUAGE; if (!isManualLanguage(value)) { throw new BadRequestException(`language must be one of ${MANUAL_LANGUAGES.join(" | ")}`); } return value; } /** A name a collective can carry; same shape rule as GroupsService. */ export function collectiveName(name: string): string { const clean = name.trim(); if (clean === "") throw new BadRequestException("name must be a non-empty string"); if (clean.length > 120) throw new BadRequestException("name must be at most 120 characters"); return clean; } /** * The collective's language rides the existing `description` column as a * trailing `lang:` line rather than a new column: P1 keeps the migration * to exactly the three objects the design names (parent_id, kind, group_rooms), * and the language is presentation metadata, not relational data. */ const LANGUAGE_TAG_PREFIX = "lang:"; function stripLanguageTag(description: string): string { const lines = description.split("\n"); while (lines.length > 0 && lines[lines.length - 1]!.trim() === "") lines.pop(); if (lines.length > 0 && lines[lines.length - 1]!.startsWith(LANGUAGE_TAG_PREFIX)) lines.pop(); return lines.join("\n"); } function encodeLanguage(description: string, language: ManualLanguage): string { const stripped = stripLanguageTag(description); return stripped === "" ? `${LANGUAGE_TAG_PREFIX}${language}` : `${stripped}\n${LANGUAGE_TAG_PREFIX}${language}`; } export function decodeLanguage(description: string | null, fallback: ManualLanguage = DEFAULT_MANUAL_LANGUAGE): ManualLanguage { if (typeof description !== "string") return fallback; const lines = description.split("\n"); for (let i = lines.length - 1; i >= 0; i -= 1) { const line = lines[i]!.trim(); if (line === "") continue; if (line.startsWith(LANGUAGE_TAG_PREFIX)) { const value = line.slice(LANGUAGE_TAG_PREFIX.length); return isManualLanguage(value) ? value : fallback; } return fallback; } return fallback; } /** The description as any client should render it: stored text minus the language tag. */ export function publicDescription(description: string | null): string { return stripLanguageTag(description ?? ""); } interface GroupRow { id: string; name: string; description: string; kind: GroupKind; parent_id: string | null; created_at: Date; member_count: string | number; } interface RoomRow { group_id: string; room_type: CollectiveRoomType; room_id: string; } function findNode(nodes: CollectiveNode[], id: string): CollectiveNode | undefined { for (const node of nodes) { if (node.id === id) return node; const found = findNode(node.children, id); if (found !== undefined) return found; } return undefined; } @Injectable() export class CollectivesService { constructor(@Inject(PROVISIONER_PORT) private readonly provisioner: ProvisionerPort) {} async createCollective(workspaceId: string, input: CreateCollectiveInput, actor: CollectiveActor): Promise { const name = collectiveName(input.name); const language = input.language; let rooms: ProvisionedRoomSet; try { rooms = await this.provisioner.provisionCollective({ workspaceId, groupId: "pending", name, language, founderUserId: actor.userId, founderMxid: null, }); } catch (err) { if (err instanceof ProvisionerUnavailableError) { throw new InternalServerErrorException(`collective rooms could not be created: ${err.message}`); } throw err; } const groupId = await withTenant(workspaceId, async (client) => { await client.query("SELECT id FROM workspaces WHERE id = $1 FOR UPDATE", [workspaceId]); let id: string; try { const res = await client.query( `INSERT INTO groups (workspace_id, name, description, kind, parent_id) VALUES ($1, $2, $3, 'collective', NULL) RETURNING id`, [workspaceId, name, encodeLanguage(input.description ?? "", language)], ); id = res.rows[0]!.id as string; } catch (err) { if (pgCode(err) === UNIQUE_VIOLATION) throw new ConflictException(`a group named "${name}" already exists`); throw err; } await this.insertFounder(client, workspaceId, id, actor.userId); await this.insertRooms(client, workspaceId, id, [ ["open", rooms.open.roomId], ["closed", rooms.closed.roomId], ["onboarding", rooms.onboarding.roomId], ]); return id; }); await appendTenantAudit(workspaceId, { actorUserId: actor.userId, actorRole: actor.role, action: "collective.created", targetType: "group", targetId: groupId, // Room IDs only — never tokens, never power-level payloads. diff: { name, language, kind: "collective", openRoom: rooms.open.roomId, closedRoom: rooms.closed.roomId }, }); const tree = await this.collectiveTree(workspaceId); const node = findNode(tree, groupId); if (node === undefined) throw new InternalServerErrorException("collective was created but is not readable in its workspace"); return node; } /** * Root-collective count + the workspace's tenancy kind, for the post-login * home decision (operator flow reorder #3). A personal workspace reports zero * by construction — it never contains collective rows — but the kind is * returned anyway so the client can label the menu honestly ("solo account") * without a second round-trip. */ async myCollectiveCount(workspaceId: string): Promise<{ collectives: number; workspaceKind: "shared" | "personal" }> { return withTenant(workspaceId, async (client) => { const rows = await client.query<{ n: string }>( "SELECT count(*)::text AS n FROM groups WHERE workspace_id = $1 AND kind = 'collective'", [workspaceId], ); const kind = await client.query<{ kind: "shared" | "personal" }>("SELECT kind FROM workspaces WHERE id = $1", [workspaceId]); return { collectives: Number(rows.rows[0]?.n ?? 0), workspaceKind: kind.rows[0]?.kind ?? "shared" }; }); } /** * The collective tree for the workspace: collectives as roots, subcollectives * nested, every node carrying its room set. 'plain' groups are sharing * scopes, not participation units, so they are not part of this tree. */ async collectiveTree(workspaceId: string): Promise { const res = await withTenant(workspaceId, async (client) => { const groups = await client.query( `SELECT g.id, g.name, g.description, g.kind, g.parent_id, g.created_at, (SELECT count(*) FROM group_members gm WHERE gm.group_id = g.id) AS member_count FROM groups g WHERE g.kind IN ('collective','subcollective') ORDER BY g.created_at ASC, g.id ASC`, ); const rooms = await client.query( "SELECT group_id, room_type, room_id FROM group_rooms WHERE workspace_id = $1", [workspaceId], ); return { groups, rooms: rooms.rows as RoomRow[] }; }); const byId = new Map(); for (const row of res.groups.rows as GroupRow[]) { byId.set(row.id, { id: row.id, name: row.name, kind: row.kind, language: decodeLanguage(row.description), parentId: row.parent_id, description: publicDescription(row.description), createdAt: new Date(row.created_at).toISOString(), memberCount: Number(row.member_count), rooms: [], children: [], }); } const roomIndex = new Map(); for (const row of res.rooms) { const list = roomIndex.get(row.group_id) ?? []; list.push({ roomType: row.room_type, roomId: row.room_id }); roomIndex.set(row.group_id, list); } const roots: CollectiveNode[] = []; for (const node of byId.values()) { node.rooms = (roomIndex.get(node.id) ?? []).sort((a, b) => a.roomType.localeCompare(b.roomType)); if (node.parentId !== null && byId.has(node.parentId)) byId.get(node.parentId)!.children.push(node); else roots.push(node); } return roots.sort((a, b) => a.createdAt.localeCompare(b.createdAt)); } /** * Join-by-invite integration (P1): given an invite token, the collective * NAMES of the workspace that token points at — nothing else. * * The joiner has NO tenant context (they have not claimed yet), so the * workspace is resolved through the existing narrow SECURITY DEFINER lookup * `invite_link_for_claim(token)` (017): the caller proves possession of the * secret first, and the projection stays at names-only (no room ids, no * member counts, no descriptions). The claim itself is untouched — * InviteClaimController keeps doing exactly what it did. */ async collectivesForInviteToken(token: string): Promise<{ collectives: Array<{ id: string; name: string }> }> { // Step 1 resolves the workspace from the secret the caller proved possession // of (SECURITY DEFINER, no tenant context needed). Step 2 reads the names // inside a tenant transaction, because `groups` is FORCE RLS: without // app.workspace_id even this service would see zero rows — by design. const workspaceId = await withSystem(async (client) => { const link = await client.query("SELECT workspace_id, revoked_at, expires_at FROM invite_link_for_claim($1)", [token]); const row = link.rows[0] as { workspace_id: string; revoked_at: Date | null; expires_at: Date } | undefined; if (row === undefined) throw new NotFoundException("invite link is invalid or has expired"); if (row.revoked_at !== null) throw new GoneException("invite link has been revoked"); if (new Date(row.expires_at).getTime() <= Date.now()) throw new GoneException("invite link has expired"); return row.workspace_id; }); return withTenant(workspaceId, async (client) => { const names = await client.query( "SELECT id, name FROM groups WHERE workspace_id = $1 AND kind IN ('collective','subcollective') ORDER BY created_at ASC", [workspaceId], ); return { collectives: (names.rows as Array<{ id: string; name: string }>).map((row) => ({ id: row.id, name: row.name })), }; }); } private async insertFounder(client: PoolClient, workspaceId: string, groupId: string, userId: string): Promise { try { await client.query( `INSERT INTO group_members (workspace_id, group_id, user_id) VALUES ($1, $2, $3) ON CONFLICT (group_id, user_id) DO NOTHING`, [workspaceId, groupId, userId], ); } catch (err) { // Composite FK to memberships: the founder is not a member of this workspace. if (pgCode(err) === FK_VIOLATION) throw new BadRequestException("founder is not a member of this workspace"); throw err; } } private async insertRooms( client: PoolClient, workspaceId: string, groupId: string, rooms: ReadonlyArray, ): Promise { for (const [roomType, roomId] of rooms) { await client.query("INSERT INTO group_rooms (workspace_id, group_id, room_type, room_id) VALUES ($1, $2, $3, $4)", [ workspaceId, groupId, roomType, roomId, ]); } } }