glass beadian
18afff5e49
The Files tab was the only payload path not on object storage, violating the 2026-10-06 directive (object storage is the ONLY media I/O path) — and oCIS's WebDAV surface is currently 500ing server-side, which made every upload fail. Uploads now seal + dedupe into R2 via StorageService (SPENC1, content- addressed); downloads read R2 when storage_key is set, falling back to oCIS for legacy rows. Share-into-library copies oCIS Drive sources into R2 and records the real sha256 instead of a placeholder. Migration 041 adds chat_media_archive.storage_key.
481 lines
19 KiB
TypeScript
481 lines
19 KiB
TypeScript
/**
|
||
* "Share with collective" service — the member-gated push of a caller's own
|
||
* content into a collective's library (oCIS group folder + chat_media_archive
|
||
* + group_file_shares) and/or its OPEN room (provisioner bot post).
|
||
*
|
||
* ACL ordering follows the getFile pattern exactly, so the failure a stranger
|
||
* sees is indistinguishable from any other unknown collective:
|
||
*
|
||
* 1. resolve the collective row (workspace-scoped; absent => NotFound)
|
||
* 2. member gate: group_members row for (collective, caller) — non-members
|
||
* (and non-Owner/Admin) get Forbidden, BEFORE any room identity or
|
||
* folder name is disclosed
|
||
* 3. only then read group_rooms (which room is 'open' vs 'closed') and
|
||
* perform the copy / post
|
||
*
|
||
* Every item is settled independently: one bad path never blocks the rest, and
|
||
* the response carries a per-item result the UI renders row by row.
|
||
*/
|
||
import { BadRequestException, ForbiddenException, Inject, Injectable, NotFoundException, Optional } from "@nestjs/common";
|
||
import { safeFileName } from "@tactical-media-broadcast-hub/core";
|
||
import type { ManualLanguage } from "@tactical-media-broadcast-hub/core";
|
||
import type { PoolClient } from "pg";
|
||
import { withTenant } from "../db/client.js";
|
||
import { isUuid } from "../http/validation.js";
|
||
import { decodeLanguage } from "../collectives/collectives.service.js";
|
||
import { appendTenantAudit } from "../audit/audit.service.js";
|
||
import {
|
||
COLLECTIVE_ROOM_POSTER,
|
||
davSegments,
|
||
e2eeRefusalMessage,
|
||
mimeClassOf,
|
||
noLibraryTargetMessage,
|
||
openRoomNoticeContent,
|
||
wantsLibrary,
|
||
wantsOpenRoom,
|
||
type CollectiveRoomPoster,
|
||
type CollectiveShareInput,
|
||
type CollectiveShareItem,
|
||
type CollectiveShareItemResult,
|
||
type CollectiveShareResult,
|
||
type ShareErrorCode,
|
||
} from "./collective-share.js";
|
||
import { OcisWebdavClient, ocisConfigFromEnv, type FetchLike } from "./ocis-webdav.js";
|
||
import { S3Storage, s3ConfigFromEnv, type StorageService } from "../storage/storage.js";
|
||
|
||
interface CollectiveRow {
|
||
id: string;
|
||
name: string;
|
||
description: string | null;
|
||
kind: string;
|
||
}
|
||
|
||
interface RoomRow {
|
||
room_type: string;
|
||
room_id: string;
|
||
}
|
||
|
||
/** The ACL-checked slice of a collective this service acts on. */
|
||
interface ResolvedCollective {
|
||
id: string;
|
||
name: string;
|
||
language: ManualLanguage;
|
||
openRoomId: string | null;
|
||
}
|
||
|
||
export const COLLECTIVE_SHARE_DEPENDENCIES: unique symbol = Symbol("COLLECTIVE_SHARE_DEPENDENCIES");
|
||
|
||
@Injectable()
|
||
export class CollectiveShareService {
|
||
private client: OcisWebdavClient | null = null;
|
||
|
||
constructor(
|
||
@Optional() @Inject(COLLECTIVE_ROOM_POSTER) private readonly poster?: CollectiveRoomPoster,
|
||
@Optional() @Inject(COLLECTIVE_SHARE_DEPENDENCIES) private readonly deps?: { fetcher?: FetchLike; storage?: StorageService },
|
||
) {}
|
||
|
||
private store: StorageService | null = null;
|
||
|
||
/** R2 object storage — the library payload store (041). */
|
||
private storage(): StorageService {
|
||
if (this.store === null) {
|
||
this.store = this.deps?.storage ?? new S3Storage(s3ConfigFromEnv());
|
||
}
|
||
return this.store;
|
||
}
|
||
|
||
private libraryUrlFor(resourceId: string): string {
|
||
const base = (process.env.FILES_PUBLIC_BASE ?? "").replace(/\/+$/, "");
|
||
return base === "" ? resourceId : `${base}/f/${encodeURIComponent(resourceId)}`;
|
||
}
|
||
|
||
/** Lazily bound: the oCIS env may be absent in dev/test contexts. */
|
||
private webdav(): OcisWebdavClient {
|
||
if (this.client === null) {
|
||
const fetcher: FetchLike = this.deps?.fetcher ?? ((url, init) => fetch(url, init));
|
||
this.client = new OcisWebdavClient(ocisConfigFromEnv(), fetcher);
|
||
}
|
||
return this.client;
|
||
}
|
||
|
||
private roomPoster(): CollectiveRoomPoster {
|
||
if (this.poster === undefined) {
|
||
// Distinguishable at a glance in a stack trace: a module wiring bug, not a runtime condition.
|
||
throw new NotFoundException("collective room poster is not wired");
|
||
}
|
||
return this.poster;
|
||
}
|
||
|
||
/**
|
||
* ACL + room resolution (steps 1–2 of the ordering above). Strangers and
|
||
* non-members learn nothing beyond "no such collective".
|
||
*/
|
||
private async resolveCollective(workspaceId: string, collectiveId: string, userId: string, role: string): Promise<ResolvedCollective> {
|
||
if (!isUuid(collectiveId)) throw new BadRequestException("collectiveId must be a UUID");
|
||
const { row, rooms, isMember } = await withTenant(workspaceId, async (client) => {
|
||
const groupResult = await client.query<CollectiveRow>(
|
||
"SELECT id, name, description, kind FROM groups WHERE id = $1 AND kind IN ('collective','subcollective')",
|
||
[collectiveId],
|
||
);
|
||
const group = groupResult.rows[0];
|
||
const memberResult = await client.query(
|
||
"SELECT 1 FROM group_members WHERE group_id = $1 AND user_id = $2 LIMIT 1",
|
||
[collectiveId, userId],
|
||
);
|
||
const roomResult = group === undefined
|
||
? { rows: [] as RoomRow[] }
|
||
: await client.query<RoomRow>("SELECT room_type, room_id FROM group_rooms WHERE group_id = $1", [collectiveId]);
|
||
return { row: group, rooms: roomResult.rows, isMember: (memberResult.rowCount ?? 0) > 0 };
|
||
});
|
||
// Ordering: 404 first (the row must exist AND belong to this workspace's
|
||
// tree — RLS has already filtered other tenants), then the member gate.
|
||
if (row === undefined) throw new NotFoundException(`collective ${collectiveId} not found`);
|
||
if (!isMember && !WORKSPACE_WIDE_ROLES.includes(role)) {
|
||
throw new ForbiddenException("you are not a member of this collective");
|
||
}
|
||
const openRoom = rooms.find((room) => room.room_type === "open");
|
||
return {
|
||
id: row.id,
|
||
name: row.name,
|
||
language: decodeLanguage(row.description),
|
||
openRoomId: openRoom === undefined ? null : openRoom.room_id,
|
||
};
|
||
}
|
||
|
||
/**
|
||
* Share items into a collective. Library copies are indexed + linked per
|
||
* item; open-room notices are posted per item; 'both' links the notice to
|
||
* the library copy that was just made.
|
||
*/
|
||
async shareWithCollective(
|
||
workspaceId: string,
|
||
collectiveId: string,
|
||
actor: { userId: string; role: string; username: string },
|
||
input: CollectiveShareInput,
|
||
): Promise<CollectiveShareResult> {
|
||
const collective = await this.resolveCollective(workspaceId, collectiveId, actor.userId, actor.role);
|
||
const doLibrary = wantsLibrary(input.target);
|
||
const doRoom = wantsOpenRoom(input.target);
|
||
|
||
if (doRoom && collective.openRoomId === null) {
|
||
// A collective without an open room cannot receive any bot post; this is
|
||
// a whole-request refusal, not per-item noise.
|
||
throw new BadRequestException(
|
||
collective.language === "es"
|
||
? "Este colectivo no tiene sala abierta configurada."
|
||
: "This collective has no open room configured.",
|
||
);
|
||
}
|
||
|
||
const senderMxid = `@${actor.username}:tmbh.local`;
|
||
const results: CollectiveShareItemResult[] = [];
|
||
for (const [index, item] of input.items.entries()) {
|
||
results.push(await this.shareOne(workspaceId, collective, actor, input, item, index, senderMxid, doLibrary, doRoom));
|
||
}
|
||
|
||
await appendTenantAudit(workspaceId, {
|
||
actorUserId: actor.userId,
|
||
actorRole: actor.role,
|
||
action: "collective.shared",
|
||
targetType: "group",
|
||
targetId: collective.id,
|
||
// Names + outcome counts only: no paths, no room ids, no note text.
|
||
diff: {
|
||
target: input.target,
|
||
collective: collective.name,
|
||
items: results.length,
|
||
libraryCopies: results.filter((r) => r.library !== null).length,
|
||
openRoomPosts: results.filter((r) => r.openRoom !== null).length,
|
||
},
|
||
});
|
||
|
||
return { collectiveId: collective.id, collectiveName: collective.name, target: input.target, items: results };
|
||
}
|
||
|
||
private async shareOne(
|
||
workspaceId: string,
|
||
collective: ResolvedCollective,
|
||
actor: { userId: string; role: string },
|
||
input: CollectiveShareInput,
|
||
item: CollectiveShareItem,
|
||
index: number,
|
||
senderMxid: string,
|
||
doLibrary: boolean,
|
||
doRoom: boolean,
|
||
): Promise<CollectiveShareItemResult> {
|
||
if (item.kind === "peertube-video" && doLibrary && !doRoom) {
|
||
// A video has no library bytes to copy; the target combination is invalid for it.
|
||
return {
|
||
index,
|
||
kind: item.kind,
|
||
title: item.title,
|
||
library: null,
|
||
openRoom: null,
|
||
error: { code: "no-library-target", message: noLibraryTargetMessage(collective.language) },
|
||
};
|
||
}
|
||
|
||
let library: CollectiveShareItemResult["library"] = null;
|
||
if (doLibrary && item.kind === "ocis-resource") {
|
||
const copied = await this.copyIntoLibrary(workspaceId, collective, actor, item, input.folder, senderMxid);
|
||
if ("error" in copied) {
|
||
return { index, kind: item.kind, title: copied.title, library: null, openRoom: null, error: copied.error };
|
||
}
|
||
library = copied.result;
|
||
}
|
||
|
||
let openRoom: CollectiveShareItemResult["openRoom"] = null;
|
||
let failure: { code: ShareErrorCode; message: string } | null = null;
|
||
if (doRoom) {
|
||
const title = item.kind === "peertube-video" ? item.title : library?.name ?? titleFromPath(item.path);
|
||
const url = item.kind === "peertube-video"
|
||
? item.url
|
||
: library?.url ?? libraryUrlFor(webdavSafePath(input.folder, item.path));
|
||
const posted = await this.postToOpenRoom(collective, { title, url, note: input.note, linkKind: item.kind === "peertube-video" ? "video" : "library" });
|
||
if ("error" in posted) failure = posted.error;
|
||
else openRoom = posted.result;
|
||
}
|
||
|
||
if (failure !== null && library === null) {
|
||
return { index, kind: item.kind, title: item.kind === "peertube-video" ? item.title : titleFromPath(item.path), library: null, openRoom: null, error: failure };
|
||
}
|
||
return {
|
||
index,
|
||
kind: item.kind,
|
||
title: item.kind === "peertube-video" ? item.title : library?.name ?? titleFromPath(item.path),
|
||
library,
|
||
openRoom,
|
||
error: failure,
|
||
};
|
||
}
|
||
|
||
/**
|
||
* COPY + index for one oCIS resource. Collision naming (stem-1, stem-2 …)
|
||
* is applied here; the caller's chosen destination folder is honored
|
||
* relative to the collective's group root.
|
||
*/
|
||
private async copyIntoLibrary(
|
||
workspaceId: string,
|
||
collective: ResolvedCollective,
|
||
actor: { userId: string; role: string },
|
||
item: { kind: "ocis-resource"; path: string },
|
||
folder: readonly string[],
|
||
senderMxid: string,
|
||
): Promise<
|
||
| { title: string; result: { itemId: string; name: string; url: string } }
|
||
| { title: string; error: { code: ShareErrorCode; message: string } }
|
||
> {
|
||
const segments = davSegments(item.path, "path");
|
||
const stat = await this.webdav().statPath(segments).catch(() => null);
|
||
if (stat === null || !stat.exists) {
|
||
return {
|
||
title: titleFromPath(item.path),
|
||
error: {
|
||
code: "source-not-found",
|
||
message: collective.language === "es"
|
||
? `No se encontró el archivo de origen (${item.path}).`
|
||
: `The source file was not found (${item.path}).`,
|
||
},
|
||
};
|
||
}
|
||
|
||
const name = safeFileName(segments[segments.length - 1] ?? "attachment");
|
||
// oCIS stays the Drive source; the library destination is R2 (041).
|
||
const source = await this.webdav().download(segments.join("/")).catch(() => null);
|
||
if (source === null || source.stream === null) {
|
||
return {
|
||
title: titleFromPath(item.path),
|
||
error: {
|
||
code: "source-not-found",
|
||
message: collective.language === "es"
|
||
? `No se encontró el archivo de origen (${item.path}).`
|
||
: `The source file was not found (${item.path}).`,
|
||
},
|
||
};
|
||
}
|
||
const chunks: Buffer[] = [];
|
||
for await (const chunk of source.stream as unknown as AsyncIterable<unknown>) {
|
||
chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk as Uint8Array));
|
||
}
|
||
const bytes = Buffer.concat(chunks);
|
||
const stored = await this.storage().putContent(bytes, source.contentType ?? "application/octet-stream").catch(() => null);
|
||
if (stored === null) {
|
||
return {
|
||
title: name,
|
||
error: {
|
||
code: "copy-failed",
|
||
message: collective.language === "es"
|
||
? "La copia en la biblioteca del colectivo falló."
|
||
: "Copying into the collective's library failed.",
|
||
},
|
||
};
|
||
}
|
||
|
||
const indexed = await this.indexCopy(workspaceId, collective, actor, {
|
||
folder,
|
||
name,
|
||
storageKey: stored.key,
|
||
sha256: stored.hash,
|
||
sizeBytes: stored.sizeBytes,
|
||
contentType: source.contentType ?? "application/octet-stream",
|
||
sourcePath: item.path,
|
||
senderMxid,
|
||
});
|
||
if ("error" in indexed) return { title: name, error: indexed.error };
|
||
return { title: name, result: indexed.result };
|
||
}
|
||
|
||
/**
|
||
* Index the copy in chat_media_archive (origin 'upload' — the caller put it
|
||
* there themselves) and mint the members library link row in
|
||
* group_file_shares so the collective's files index lists it.
|
||
*/
|
||
private async indexCopy(
|
||
workspaceId: string,
|
||
collective: ResolvedCollective,
|
||
actor: { userId: string; role: string },
|
||
copy: {
|
||
folder: readonly string[];
|
||
name: string;
|
||
storageKey: string;
|
||
sha256: string;
|
||
sizeBytes: number;
|
||
contentType: string;
|
||
sourcePath: string;
|
||
senderMxid: string;
|
||
},
|
||
): Promise<{ result: { itemId: string; name: string; url: string } } | { error: { code: ShareErrorCode; message: string } }> {
|
||
const mime = copy.contentType === "" ? "application/octet-stream" : copy.contentType;
|
||
const libraryUrl = this.libraryUrlFor(copy.storageKey);
|
||
const folderPath = copy.folder.length === 0 ? "" : `${copy.folder.join("/")}/`;
|
||
const origin = {
|
||
kind: "upload",
|
||
sharedFrom: copy.sourcePath,
|
||
sharedInto: `hub-groups/${collective.id}/${folderPath}${copy.name}`,
|
||
};
|
||
const inserted = await withTenant(workspaceId, async (client: PoolClient) => {
|
||
const row = await client.query<{ id: string }>(
|
||
`INSERT INTO chat_media_archive
|
||
(workspace_id, group_id, room_id, event_id, origin, sender_mxid, matrix_event_json,
|
||
original_name, safe_name, mime, mime_class, byte_size, sha256, ocis_resource_id,
|
||
ocis_resource_version, created_at, indexed_at, storage_key)
|
||
VALUES ($1,$2,NULL,NULL,'upload',$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,now(),now(),$13)
|
||
RETURNING id`,
|
||
[
|
||
workspaceId,
|
||
collective.id,
|
||
copy.senderMxid,
|
||
JSON.stringify(origin),
|
||
copy.name,
|
||
copy.name,
|
||
mime,
|
||
mimeClassOf(mime),
|
||
copy.sizeBytes,
|
||
copy.sha256,
|
||
copy.storageKey,
|
||
null,
|
||
copy.storageKey,
|
||
],
|
||
);
|
||
return row.rows[0]?.id;
|
||
}).catch(() => undefined);
|
||
if (inserted === undefined) {
|
||
return {
|
||
error: {
|
||
code: "index-failed",
|
||
message: collective.language === "es"
|
||
? "La copia se realizó pero no se pudo registrar en el índice."
|
||
: "The copy succeeded but indexing it failed.",
|
||
},
|
||
};
|
||
}
|
||
return { result: { itemId: inserted, name: copy.name, url: libraryUrl } };
|
||
}
|
||
|
||
/**
|
||
* JOIN-if-not-in-room, then one m.text notice. A closed room is refused
|
||
* before any Matrix call: this service is only ever handed the OPEN room
|
||
* id, and the refusal message is the domain text the UI shows verbatim.
|
||
*/
|
||
private async postToOpenRoom(
|
||
collective: ResolvedCollective,
|
||
notice: { title: string; url: string; note: string | null; linkKind: "library" | "video" },
|
||
): Promise<{ result: { roomId: string; eventId: string } } | { error: { code: ShareErrorCode; message: string } }> {
|
||
if (collective.openRoomId === null) {
|
||
// Only ever reached for a collective whose room set is broken; the
|
||
// E2EE-shaped refusal text is the clearest honest answer.
|
||
return { error: { code: "room-missing", message: e2eeRefusalMessage(collective.language) } };
|
||
}
|
||
const content = openRoomNoticeContent({
|
||
language: collective.language,
|
||
title: notice.title,
|
||
url: notice.url,
|
||
note: notice.note,
|
||
linkKind: notice.linkKind,
|
||
});
|
||
const poster = this.roomPoster();
|
||
const txnId = `hub-share-${collective.id}-${Date.now()}-${Math.floor(Math.random() * 1e6)}`;
|
||
try {
|
||
await poster.joinRoom(collective.openRoomId);
|
||
const sent = await poster.sendRoomMessage(collective.openRoomId, txnId, content);
|
||
return { result: { roomId: collective.openRoomId, eventId: sent.eventId } };
|
||
} catch {
|
||
return {
|
||
error: {
|
||
code: "matrix-unavailable",
|
||
message: collective.language === "es"
|
||
? "No se pudo publicar en la sala abierta del colectivo."
|
||
: "Posting into the collective's open room failed.",
|
||
},
|
||
};
|
||
}
|
||
}
|
||
|
||
/** The collectives a caller may share into: members of, workspace-wide for Owner/Admin. */
|
||
async myCollectives(workspaceId: string, userId: string, role: string): Promise<Array<{ id: string; name: string }>> {
|
||
return withTenant(workspaceId, async (client) => {
|
||
if (WORKSPACE_WIDE_ROLES.includes(role)) {
|
||
const all = await client.query<CollectiveRow>(
|
||
"SELECT id, name FROM groups WHERE kind IN ('collective','subcollective') ORDER BY name ASC",
|
||
);
|
||
return all.rows.map((row) => ({ id: row.id, name: row.name }));
|
||
}
|
||
const mine = await client.query<CollectiveRow>(
|
||
`SELECT g.id, g.name FROM groups g
|
||
WHERE g.kind IN ('collective','subcollective')
|
||
AND EXISTS (SELECT 1 FROM group_members gm WHERE gm.group_id = g.id AND gm.user_id = $1)
|
||
ORDER BY g.name ASC`,
|
||
[userId],
|
||
);
|
||
return mine.rows.map((row) => ({ id: row.id, name: row.name }));
|
||
});
|
||
}
|
||
}
|
||
|
||
const WORKSPACE_WIDE_ROLES: readonly string[] = ["Owner", "Admin"];
|
||
|
||
/** Last path segment as a display title. */
|
||
function titleFromPath(path: string): string {
|
||
const segment = path.split("/").filter((part) => part !== "").at(-1) ?? path;
|
||
return segment;
|
||
}
|
||
|
||
/**
|
||
* A server-side COPY never streams bytes through the API, so no digest is
|
||
* computed here. The column is NOT NULL, so the row stores a deterministic
|
||
* marker derived from the stable resource id — visibly not a payload digest
|
||
* (it is not 64 lowercase hex) and never presented as one.
|
||
*/
|
||
function digestPlaceholder(resourceId: string): string {
|
||
return `copied:${resourceId}`;
|
||
}
|
||
|
||
function webdavSafePath(folder: readonly string[], path: string): string {
|
||
const segments = davSegments(path, "path");
|
||
return folder.length === 0 ? segments.join("/") : `${folder.join("/")}/${segments.join("/")}`;
|
||
}
|
||
|
||
function libraryUrlFor(path: string): string {
|
||
const base = process.env.FILES_PUBLIC_BASE ?? "";
|
||
return base === "" ? path : `${base.replace(/\/+$/, "")}/f/${encodeURIComponent(path)}`;
|
||
}
|