glass beadian
aa5e262f3a
Self-hosted multi-platform broadcast newsroom for activist collectives: publishing pipeline with group approval and DST-proof scheduling, TweetDeck-style column wall with hostile-content-safe rendering, live control room with Auto Director, bridged chat governance with polls and quorum, distributed sandboxed transcode workers, encrypted content-addressed archive, monthly analytics. MIT licensed. See README.md and docs/.
460 lines
20 KiB
TypeScript
460 lines
20 KiB
TypeScript
import { BadRequestException, ConflictException, Inject, Injectable, NotFoundException, Optional, ServiceUnavailableException } from "@nestjs/common";
|
|
import { randomUUID } from "node:crypto";
|
|
import type { PoolClient } from "pg";
|
|
import { withTenant } from "../db/client.js";
|
|
import { isUuid, requireString } from "../http/validation.js";
|
|
import { appendTenantAudit } from "../audit/audit.service.js";
|
|
import { logger } from "../telemetry/logger.js";
|
|
import {
|
|
dedupeSyntactic,
|
|
IMPORT_ROW_LIMIT,
|
|
InvalidRowError,
|
|
normalizeImportRow,
|
|
parseImportBody,
|
|
type ImportRowKind,
|
|
type NormalizedRow,
|
|
} from "./import-normalize.js";
|
|
import { egressFailure, IMPORT_HOST_CLASS, importEgress, type ImportEgress } from "./import-egress.js";
|
|
import {
|
|
PeerTubeResolveError,
|
|
resolvePeerTubeSource,
|
|
type ResolvedPeerTubeSource,
|
|
} from "../../../../packages/connector-peertube/src/resolve.js";
|
|
|
|
/** Resolution cache: 10 minutes for live-capable rows, 1 minute otherwise. */
|
|
export const RESOLUTION_CACHE_TTL_MS = 10 * 60 * 1000;
|
|
export const RESOLUTION_CACHE_SHORT_TTL_MS = 60 * 1000;
|
|
/** Per-import concurrency bound; per-host bound comes from the egress budgets. */
|
|
export const IMPORT_MAX_CONCURRENCY = 16;
|
|
/** Cap on per-entry detail rows stored in the report document. */
|
|
export const IMPORT_REPORT_ENTRY_LIMIT = 200;
|
|
|
|
export type ImportOutcome = "imported" | "duplicate" | "invalid" | "failed";
|
|
|
|
export interface ImportReportEntry {
|
|
raw: string;
|
|
key?: string;
|
|
remoteRef?: string;
|
|
outcome: ImportOutcome;
|
|
reason?: string;
|
|
message?: string;
|
|
}
|
|
|
|
export interface ImportJobRecord {
|
|
jobId: string;
|
|
workspaceId: string;
|
|
virtualChannelId: string;
|
|
connectionId: string | null;
|
|
status: "running" | "completed" | "failed" | "abandoned";
|
|
format: string;
|
|
total: number;
|
|
processed: number;
|
|
imported: number;
|
|
duplicate: number;
|
|
invalid: number;
|
|
failed: number;
|
|
entries: ImportReportEntry[];
|
|
error?: string;
|
|
createdAt: string;
|
|
updatedAt: string;
|
|
completedAt?: string;
|
|
}
|
|
|
|
/** Injected transport seam: production binds the egress client lazily. */
|
|
export type ImportResolver = (kind: ImportRowKind, name: string, host: string, egress: ImportEgress) => Promise<ResolvedPeerTubeSource>;
|
|
|
|
export interface VirtualChannelImportDependencies {
|
|
tenantTransaction: typeof withTenant;
|
|
/** Audit sink; defaults to the shared appendTenantAudit. */
|
|
audit?: (workspaceId: string, event: { actorUserId: string | null; actorRole: string; action: string; targetType: string; targetId: string; diff?: Record<string, unknown> | null }) => Promise<unknown>;
|
|
egress?: () => Promise<ImportEgress>;
|
|
resolver?: ImportResolver;
|
|
/** Unit seam: deterministic clock + scheduler. */
|
|
now?: () => number;
|
|
/** In-process resolution cache (host -> key -> resolution + expiry). */
|
|
cache?: Map<string, { value: ResolvedPeerTubeSource; expiresAt: number }>;
|
|
/** Progress-flush cadence; 0 flushes every row. */
|
|
progressEvery?: number;
|
|
}
|
|
|
|
export const VIRTUAL_CHANNEL_IMPORT_DEPENDENCIES = Symbol("VIRTUAL_CHANNEL_IMPORT_DEPENDENCIES");
|
|
|
|
interface RowResult {
|
|
outcome?: ImportOutcome;
|
|
reason?: string;
|
|
message?: string;
|
|
remoteRef?: string;
|
|
}
|
|
|
|
/**
|
|
* Bulk import of PeerTube sources into a virtual channel (WP7).
|
|
*
|
|
* The HTTP request parses + normalizes + syntactically dedupes and creates the
|
|
* job row synchronously; resolution and inserts then run in the background
|
|
* (see runImportJob). Concurrency is bounded twice: at most 16 resolutions in
|
|
* flight per import, and per host by the shared egress budget rows (drawn
|
|
* through egressFetch inside the resolver). The DB's UNIQUE
|
|
* (workspace_id, virtual_channel_id, remote_ref) is the final dedupe
|
|
* authority: INSERT ... ON CONFLICT DO NOTHING RETURNING id, and a conflict is
|
|
* counted `duplicate`.
|
|
*/
|
|
@Injectable()
|
|
export class VirtualChannelImportService {
|
|
private readonly deps: VirtualChannelImportDependencies;
|
|
/** jobId -> cancellation flag for graceful shutdown of running jobs. */
|
|
private readonly running = new Map<string, boolean>();
|
|
|
|
constructor(@Optional() @Inject(VIRTUAL_CHANNEL_IMPORT_DEPENDENCIES) dependencies?: Partial<VirtualChannelImportDependencies>) {
|
|
this.deps = {
|
|
tenantTransaction: withTenant,
|
|
audit: (workspaceId, event) => appendTenantAudit(workspaceId, event),
|
|
egress: () => importEgress(),
|
|
resolver: async (kind, name, host, egress) =>
|
|
resolvePeerTubeSource(kind, name, host, {
|
|
fetch: (url, init) => egress.egressFetch(url, init, { hostClass: IMPORT_HOST_CLASS, pinnedOrigin: true }),
|
|
}),
|
|
now: Date.now,
|
|
cache: new Map(),
|
|
progressEvery: 25,
|
|
...dependencies,
|
|
};
|
|
}
|
|
|
|
/** Creates the job and schedules the background run. Returns immediately. */
|
|
async createImport(
|
|
workspaceId: string,
|
|
virtualChannelId: string,
|
|
input: { contentType?: string; body: string; connectionId?: string; actor: { userId: string | null; role: string } },
|
|
): Promise<ImportJobRecord> {
|
|
const virtualChannel = parseId(virtualChannelId, "virtualChannelId");
|
|
const connectionId = input.connectionId === undefined || input.connectionId === ""
|
|
? undefined
|
|
: isUuid(input.connectionId) ? input.connectionId : (() => { throw new BadRequestException("connectionId must be a UUID"); })();
|
|
const body = requireString(input.body, "body");
|
|
if (body.length > 2_000_000) throw new BadRequestException("body exceeds 2 MB");
|
|
const parsed = parseImportBody(input.contentType, body);
|
|
if (parsed.rows.length > IMPORT_ROW_LIMIT) {
|
|
throw new BulkTooLargeException(parsed.rows.length, IMPORT_ROW_LIMIT);
|
|
}
|
|
if (parsed.rows.length === 0) throw new BadRequestException("import body contains no rows");
|
|
|
|
const created = await this.deps.tenantTransaction(workspaceId, async (client) => {
|
|
const channel = await client.query("SELECT id FROM virtual_channels WHERE workspace_id = $1 AND id = $2", [workspaceId, virtualChannel]);
|
|
if (channel.rowCount !== 1) throw new NotFoundException("virtual channel not found");
|
|
// Default host comes from the connection's instance (the only provider
|
|
// this slice resolves); no connection means bare handles are invalid.
|
|
let defaultHost: string | undefined;
|
|
if (connectionId !== undefined) {
|
|
const connection = await client.query<{ fields: Record<string, string> }>(
|
|
"SELECT fields FROM connections WHERE workspace_id = $1 AND id = $2 AND disconnected_at IS NULL",
|
|
[workspaceId, connectionId],
|
|
);
|
|
if (connection.rowCount !== 1) throw new NotFoundException("connection not found");
|
|
const instanceUrl = connection.rows[0]?.fields.instanceUrl;
|
|
if (instanceUrl !== undefined) {
|
|
defaultHost = new URL(instanceUrl).hostname.toLowerCase();
|
|
}
|
|
}
|
|
const row = await client.query(
|
|
`INSERT INTO virtual_channel_imports (workspace_id, virtual_channel_id, connection_id, status, total, created_by)
|
|
VALUES ($1, $2, $3, 'running', $4, $5) RETURNING id, created_at`,
|
|
[workspaceId, virtualChannel, connectionId ?? null, parsed.rows.length, input.actor.userId],
|
|
);
|
|
return { jobId: String(row.rows[0]?.id), createdAt: row.rows[0]?.created_at as Date, defaultHost };
|
|
});
|
|
|
|
// Background execution: never awaited by the request. Failures land on the
|
|
// job row (status failed + error), not on the caller.
|
|
const execution = this.execute(workspaceId, created.jobId, virtualChannel, connectionId, created.defaultHost, parsed.rows, input.actor)
|
|
.catch((error: unknown) => logger.error({ err: String(error), jobId: created.jobId }, "virtual channel import crashed"));
|
|
this.running.set(created.jobId, true);
|
|
void execution;
|
|
|
|
return {
|
|
jobId: created.jobId,
|
|
workspaceId,
|
|
virtualChannelId: virtualChannel,
|
|
connectionId: connectionId ?? null,
|
|
status: "running",
|
|
format: parsed.format,
|
|
total: parsed.rows.length,
|
|
processed: 0,
|
|
imported: 0,
|
|
duplicate: 0,
|
|
invalid: 0,
|
|
failed: 0,
|
|
entries: [],
|
|
createdAt: created.createdAt.toISOString(),
|
|
updatedAt: created.createdAt.toISOString(),
|
|
};
|
|
}
|
|
|
|
/** Job status + report; a stale `running` row is reaped to `abandoned`. */
|
|
async getImport(workspaceId: string, jobId: string): Promise<ImportJobRecord> {
|
|
if (!isUuid(jobId)) throw new BadRequestException("jobId must be a UUID");
|
|
return this.deps.tenantTransaction(workspaceId, async (client) => {
|
|
const found = await client.query("SELECT * FROM virtual_channel_imports WHERE workspace_id = $1 AND id = $2", [workspaceId, jobId]);
|
|
const row = found.rows[0] as ImportRow | undefined;
|
|
if (row === undefined) throw new NotFoundException("import job not found");
|
|
if (row.status === "running" && !this.running.has(row.id)
|
|
&& Date.now() - row.updated_at.getTime() > 5 * 60_000) {
|
|
await client.query(
|
|
"UPDATE virtual_channel_imports SET status = 'abandoned', error = 'job runner stopped', updated_at = now() WHERE id = $1 AND status = 'running'",
|
|
[row.id],
|
|
);
|
|
row.status = "abandoned";
|
|
}
|
|
return toRecord(row);
|
|
});
|
|
}
|
|
|
|
private async execute(
|
|
workspaceId: string,
|
|
jobId: string,
|
|
virtualChannelId: string,
|
|
connectionId: string | undefined,
|
|
defaultHost: string | undefined,
|
|
rawRows: string[],
|
|
actor: { userId: string | null; role: string },
|
|
): Promise<void> {
|
|
try {
|
|
const egress = await this.loadEgress();
|
|
// Stage 1: normalize + syntactic validation (egress policy for URLs, no
|
|
// network) and pre-resolution dedupe. Rows still count toward `total`.
|
|
const normalized: NormalizedRow[] = [];
|
|
const invalid: RowResult[] = [];
|
|
for (const raw of rawRows) {
|
|
try {
|
|
normalized.push(normalizeImportRow(raw, defaultHost, egress.policy));
|
|
} catch (error) {
|
|
if (error instanceof InvalidRowError) invalid.push({ outcome: "invalid", reason: error.reason, message: error.message });
|
|
else invalid.push({ outcome: "failed", reason: "normalize_error", message: "Row could not be processed" });
|
|
}
|
|
}
|
|
// Pre-resolution dedupe on the syntactic key: first occurrence wins,
|
|
// every later spelling of the same key counts duplicate.
|
|
const { unique } = dedupeSyntactic(normalized);
|
|
const syntacticDuplicates: Array<{ row: NormalizedRow }> = [];
|
|
const seenKeys = new Set<string>();
|
|
for (const row of normalized) {
|
|
if (seenKeys.has(row.key)) syntacticDuplicates.push({ row });
|
|
else seenKeys.add(row.key);
|
|
}
|
|
|
|
// Stage 2: resolve the unique survivors with bounded concurrency; the
|
|
// per-host bound is the egress budget drawn inside egressFetch.
|
|
const resolved: Array<{ row: NormalizedRow; result?: ResolvedPeerTubeSource; outcome?: ImportOutcome; reason?: string; message?: string }> = [];
|
|
let cursor = 0;
|
|
const worker = async (): Promise<void> => {
|
|
for (;;) {
|
|
const index = cursor++;
|
|
if (index >= unique.length) return;
|
|
const row = unique[index]!;
|
|
const cached = this.cacheGet(row);
|
|
if (cached !== undefined) {
|
|
resolved[index] = { row, result: cached };
|
|
continue;
|
|
}
|
|
try {
|
|
const resolve = this.deps.resolver;
|
|
if (resolve === undefined) throw new Error("import resolver is not configured");
|
|
const result = await resolve(row.kind, row.name, row.host, egress);
|
|
this.cachePut(row, result);
|
|
resolved[index] = { row, result };
|
|
} catch (error) {
|
|
if (error instanceof PeerTubeResolveError) {
|
|
resolved[index] = { row, outcome: error.reason === "not_found" ? "invalid" : "failed", reason: error.reason, message: error.message };
|
|
} else {
|
|
const mapped = egressFailure(error, egress.isEgressError);
|
|
resolved[index] = { row, outcome: mapped.reason === "not_found" ? "invalid" : "failed", reason: mapped.reason, message: mapped.message };
|
|
}
|
|
}
|
|
}
|
|
};
|
|
const lanes = Math.min(IMPORT_MAX_CONCURRENCY, Math.max(1, unique.length));
|
|
await Promise.all(Array.from({ length: lanes }, worker));
|
|
|
|
// Stage 3: post-resolution dedupe on remote_ref, then insert. The DB
|
|
// constraint stays authoritative: ON CONFLICT DO NOTHING RETURNING id,
|
|
// and a conflict (same remote identity reached two ways) is duplicate.
|
|
const report: ImportReportEntry[] = [];
|
|
const counters = { imported: 0, duplicate: 0, invalid: 0, failed: 0 };
|
|
const push = (entry: ImportReportEntry): void => {
|
|
if (report.length < IMPORT_REPORT_ENTRY_LIMIT) report.push(entry);
|
|
};
|
|
for (const bad of invalid) {
|
|
counters[bad.outcome ?? "invalid"] += 1;
|
|
push({ raw: "", outcome: bad.outcome ?? "invalid", reason: bad.reason, message: bad.message });
|
|
}
|
|
for (const dup of syntacticDuplicates) {
|
|
counters.duplicate += 1;
|
|
push({ raw: dup.row.raw, key: dup.row.key, outcome: "duplicate", reason: "duplicate_input" });
|
|
}
|
|
const seenRefs = new Set<string>();
|
|
const toInsert: Array<{ row: NormalizedRow; result: ResolvedPeerTubeSource }> = [];
|
|
for (const entry of resolved) {
|
|
if (entry === undefined) continue;
|
|
if (entry.outcome !== undefined) {
|
|
counters[entry.outcome] += 1;
|
|
push({ raw: entry.row.raw, key: entry.row.key, outcome: entry.outcome, reason: entry.reason, message: entry.message });
|
|
continue;
|
|
}
|
|
const result = entry.result!;
|
|
if (seenRefs.has(result.remoteRef)) {
|
|
counters.duplicate += 1;
|
|
push({ raw: entry.row.raw, key: entry.row.key, remoteRef: result.remoteRef, outcome: "duplicate", reason: "duplicate_remote_ref" });
|
|
continue;
|
|
}
|
|
seenRefs.add(result.remoteRef);
|
|
toInsert.push({ row: entry.row, result });
|
|
}
|
|
|
|
const inserted = await this.deps.tenantTransaction(workspaceId, async (client) => {
|
|
let imported = 0;
|
|
let duplicate = 0;
|
|
for (const { row, result } of toInsert) {
|
|
const label = sanitizeLabel(result.label ?? row.label);
|
|
const insert = await client.query(
|
|
`INSERT INTO virtual_channel_sources
|
|
(workspace_id, virtual_channel_id, kind, label, remote_ref, hls_url, connection_id)
|
|
VALUES ($1, $2, 'hls', $3, $4, NULLIF($5, ''), $6)
|
|
ON CONFLICT (workspace_id, virtual_channel_id, remote_ref) DO NOTHING
|
|
RETURNING id`,
|
|
[workspaceId, virtualChannelId, label, result.remoteRef, result.playbackHlsUrl ?? "", connectionId ?? null],
|
|
);
|
|
if (insert.rowCount === 1) {
|
|
imported += 1;
|
|
push({ raw: row.raw, key: row.key, remoteRef: result.remoteRef, outcome: "imported" });
|
|
} else {
|
|
duplicate += 1;
|
|
push({ raw: row.raw, key: row.key, remoteRef: result.remoteRef, outcome: "duplicate", reason: "already_imported" });
|
|
}
|
|
}
|
|
return { imported, duplicate };
|
|
});
|
|
counters.imported += inserted.imported;
|
|
counters.duplicate += inserted.duplicate;
|
|
|
|
const failedCounters = counters;
|
|
await this.deps.tenantTransaction(workspaceId, (client) =>
|
|
client.query(
|
|
`UPDATE virtual_channel_imports
|
|
SET status = 'completed', processed = total, imported = $3, duplicate = $4,
|
|
invalid = $5, failed = $6, entries = $7, completed_at = now(), updated_at = now()
|
|
WHERE workspace_id = $1 AND id = $2`,
|
|
[workspaceId, jobId, failedCounters.imported, failedCounters.duplicate, failedCounters.invalid, failedCounters.failed, JSON.stringify(report)],
|
|
));
|
|
await (this.deps.audit ?? noopAudit)(workspaceId, {
|
|
actorUserId: actor.userId, actorRole: actor.role,
|
|
action: "virtualchannel.import.completed", targetType: "virtual_channel_import", targetId: jobId,
|
|
diff: { channelId: virtualChannelId, connectionId, ...failedCounters },
|
|
}).catch(() => undefined);
|
|
} catch (error) {
|
|
const message = error instanceof Error ? error.message : "import failed";
|
|
await this.deps.tenantTransaction(workspaceId, (client) =>
|
|
client.query(
|
|
`UPDATE virtual_channel_imports SET status = 'failed', error = $3, updated_at = now(), completed_at = now()
|
|
WHERE workspace_id = $1 AND id = $2 AND status = 'running'`,
|
|
[workspaceId, jobId, message],
|
|
)).catch(() => undefined);
|
|
await (this.deps.audit ?? noopAudit)(workspaceId, {
|
|
actorUserId: actor.userId, actorRole: actor.role,
|
|
action: "virtualchannel.import.failed", targetType: "virtual_channel_import", targetId: jobId,
|
|
diff: { channelId: virtualChannelId, error: message },
|
|
}).catch(() => undefined);
|
|
} finally {
|
|
this.running.delete(jobId);
|
|
}
|
|
}
|
|
|
|
private async loadEgress(): Promise<ImportEgress> {
|
|
const load = this.deps.egress;
|
|
if (load === undefined) throw new ServiceUnavailableException("import egress is not configured");
|
|
return await load();
|
|
}
|
|
|
|
private cacheGet(row: NormalizedRow): ResolvedPeerTubeSource | undefined {
|
|
const cache = this.deps.cache;
|
|
if (cache === undefined) return undefined;
|
|
const entry = cache.get(`${row.host}|${row.key}`);
|
|
if (entry === undefined) return undefined;
|
|
if (entry.expiresAt <= (this.deps.now?.() ?? Date.now())) {
|
|
cache.delete(`${row.host}|${row.key}`);
|
|
return undefined;
|
|
}
|
|
return entry.value;
|
|
}
|
|
|
|
private cachePut(row: NormalizedRow, result: ResolvedPeerTubeSource): void {
|
|
const cache = this.deps.cache;
|
|
if (cache === undefined) return;
|
|
const now = this.deps.now?.() ?? Date.now();
|
|
const ttl = row.kind === "live" ? RESOLUTION_CACHE_TTL_MS : RESOLUTION_CACHE_SHORT_TTL_MS;
|
|
cache.set(`${row.host}|${row.key}`, { value: result, expiresAt: now + ttl });
|
|
}
|
|
}
|
|
|
|
interface ImportRow {
|
|
id: string;
|
|
workspace_id: string;
|
|
status: "running" | "completed" | "failed" | "abandoned";
|
|
total: number;
|
|
processed: number;
|
|
imported: number;
|
|
duplicate: number;
|
|
invalid: number;
|
|
failed: number;
|
|
entries: ImportReportEntry[];
|
|
error: string | null;
|
|
created_at: Date;
|
|
updated_at: Date;
|
|
completed_at: Date | null;
|
|
connection_id: string | null;
|
|
virtual_channel_id: string;
|
|
}
|
|
|
|
function toRecord(row: ImportRow): ImportJobRecord {
|
|
const record: ImportJobRecord = {
|
|
jobId: row.id,
|
|
workspaceId: row.workspace_id,
|
|
virtualChannelId: row.virtual_channel_id,
|
|
connectionId: row.connection_id,
|
|
status: row.status,
|
|
format: "import",
|
|
total: row.total,
|
|
processed: row.processed,
|
|
imported: row.imported,
|
|
duplicate: row.duplicate,
|
|
invalid: row.invalid,
|
|
failed: row.failed,
|
|
entries: Array.isArray(row.entries) ? row.entries : [],
|
|
createdAt: row.created_at.toISOString(),
|
|
updatedAt: row.updated_at.toISOString(),
|
|
};
|
|
if (row.error !== null) record.error = row.error;
|
|
if (row.completed_at !== null) record.completedAt = row.completed_at.toISOString();
|
|
return record;
|
|
}
|
|
|
|
/** 413 with a machine-readable body (proposal §4.4 cap). */
|
|
export class BulkTooLargeException extends Error {
|
|
constructor(readonly received: number, readonly limit: number) {
|
|
super(`import body has ${received} rows; the limit is ${limit}`);
|
|
this.name = "BulkTooLargeException";
|
|
}
|
|
}
|
|
|
|
function parseId(value: string, field: string): string {
|
|
if (!isUuid(value)) throw new BadRequestException(`${field} must be a UUID`);
|
|
return value;
|
|
}
|
|
|
|
/** Label CHECK in 013: alnum first, 64 chars, safe charset only. */
|
|
function sanitizeLabel(label: string): string {
|
|
const cleaned = label.replace(/[^\w .\-]/g, "").trim().slice(0, 64);
|
|
return cleaned === "" ? "imported source" : cleaned;
|
|
}
|
|
|
|
const noopAudit = async (): Promise<void> => undefined;
|
|
|
|
export type { PoolClient };
|