rspace-online/modules/rtube/mod.ts

/**
 * Tube module — community video hosting & live streaming.
 *
 * Video library (Cloudflare R2 bucket), HLS live streaming,
 * and RTMP ingest support. No database — metadata from S3 listing.
 */

import { Hono } from "hono";
import { renderShell } from "../../server/shell";
import { getModuleInfoList } from "../../shared/module";
import type { RSpaceModule } from "../../shared/module";
import { verifyEncryptIDToken, extractToken } from "@encryptid/sdk/server";
import { renderLanding } from "./landing";
import { S3Client, ListObjectsV2Command, GetObjectCommand, HeadObjectCommand, PutObjectCommand } from "@aws-sdk/client-s3";

const routes = new Hono();

// ── 360split config ──
const SPLIT_360_URL = process.env.SPLIT_360_URL || "";

// ── R2 / S3 config ──
const R2_ENDPOINT = process.env.R2_ENDPOINT || "";
const R2_BUCKET = process.env.R2_BUCKET || "rtube-videos";
const R2_ACCESS_KEY_ID = process.env.R2_ACCESS_KEY_ID || "";
const R2_SECRET_ACCESS_KEY = process.env.R2_SECRET_ACCESS_KEY || "";
const R2_PUBLIC_URL = process.env.R2_PUBLIC_URL || "";

const VIDEO_EXTENSIONS = new Set([
	".mp4", ".mkv", ".webm", ".mov", ".avi", ".wmv", ".flv", ".m4v",
]);

// ── S3 Client (lazy init) ──
let s3: S3Client | null = null;
function getS3(): S3Client | null {
	if (!R2_ENDPOINT) return null;
	if (!s3) {
		s3 = new S3Client({
			region: "auto",
			endpoint: R2_ENDPOINT,
			credentials: {
				accessKeyId: R2_ACCESS_KEY_ID,
				secretAccessKey: R2_SECRET_ACCESS_KEY,
			},
		});
	}
	return s3;
}

// ── S3-compatible listing ──
async function listVideos(): Promise<Array<{ name: string; size: number; lastModified?: string }>> {
	const client = getS3();
	if (!client) {
		// Return demo data when R2 not configured
		return [
			{ name: "welcome-to-rtube.mp4", size: 12_500_000 },
			{ name: "community-meeting-2026.mp4", size: 85_000_000 },
			{ name: "workshop-recording.webm", size: 45_000_000 },
		];
	}

	try {
		const command = new ListObjectsV2Command({ Bucket: R2_BUCKET, MaxKeys: 200 });
		const response = await client.send(command);
		const items: Array<{ name: string; size: number; lastModified?: string }> = [];

		for (const obj of response.Contents || []) {
			if (!obj.Key) continue;
			const ext = obj.Key.substring(obj.Key.lastIndexOf(".")).toLowerCase();
			if (VIDEO_EXTENSIONS.has(ext)) {
				items.push({
					name: obj.Key,
					size: obj.Size || 0,
					lastModified: obj.LastModified?.toISOString(),
				});
			}
		}
		return items.sort((a, b) => a.name.localeCompare(b.name));
	} catch (e) {
		console.error("[Tube] S3 list error:", e);
		return [];
	}
}

// ── API routes ──

// GET /api/videos — list videos from bucket
routes.get("/api/videos", async (c) => {
	const videos = await listVideos();
	return c.json({ videos, r2Configured: !!R2_ENDPOINT });
});

// GET /api/v/:path — video streaming with HTTP range request support
routes.get("/api/v/*", async (c) => {
	const client = getS3();
	if (!client) return c.json({ error: "R2 not configured" }, 503);

	const path = c.req.path.replace(/^.*\/api\/v\//, "");
	if (!path) return c.json({ error: "Path required" }, 400);

	try {
		// Get object metadata for Content-Length
		const head = await client.send(new HeadObjectCommand({ Bucket: R2_BUCKET, Key: path }));
		const totalSize = head.ContentLength || 0;
		const contentType = head.ContentType || "video/mp4";
		const rangeHeader = c.req.header("range");

		if (rangeHeader) {
			// Parse range header
			const match = rangeHeader.match(/bytes=(\d+)-(\d*)/);
			if (!match) return c.json({ error: "Invalid range" }, 400);

			const start = parseInt(match[1]);
			const end = match[2] ? parseInt(match[2]) : Math.min(start + 5 * 1024 * 1024 - 1, totalSize - 1);
			const chunkSize = end - start + 1;

			const obj = await client.send(new GetObjectCommand({
				Bucket: R2_BUCKET,
				Key: path,
				Range: `bytes=${start}-${end}`,
			}));

			return new Response(obj.Body as ReadableStream, {
				status: 206,
				headers: {
					"Content-Type": contentType,
					"Content-Length": String(chunkSize),
					"Content-Range": `bytes ${start}-${end}/${totalSize}`,
					"Accept-Ranges": "bytes",
					"Cache-Control": "public, max-age=86400",
				},
			});
		}

		// No range — stream full file
		const obj = await client.send(new GetObjectCommand({ Bucket: R2_BUCKET, Key: path }));
		return new Response(obj.Body as ReadableStream, {
			headers: {
				"Content-Type": contentType,
				"Content-Length": String(totalSize),
				"Accept-Ranges": "bytes",
				"Cache-Control": "public, max-age=86400",
			},
		});
	} catch (e: any) {
		if (e.name === "NoSuchKey") return c.json({ error: "Video not found" }, 404);
		console.error("[Tube] Stream error:", e);
		return c.json({ error: "Stream failed" }, 500);
	}
});

// POST /api/videos — upload video (auth required)
routes.post("/api/videos", async (c) => {
	const authToken = extractToken(c.req.raw.headers);
	if (!authToken) return c.json({ error: "Authentication required" }, 401);
	try { await verifyEncryptIDToken(authToken); } catch { return c.json({ error: "Invalid token" }, 401); }

	const client = getS3();
	if (!client) return c.json({ error: "R2 not configured" }, 503);

	const formData = await c.req.formData();
	const file = formData.get("video") as File | null;
	if (!file) return c.json({ error: "video file required" }, 400);

	const ext = file.name.substring(file.name.lastIndexOf(".")).toLowerCase();
	if (!VIDEO_EXTENSIONS.has(ext)) return c.json({ error: `Unsupported format. Allowed: ${[...VIDEO_EXTENSIONS].join(", ")}` }, 400);

	const key = formData.get("path")?.toString() || file.name;
	const buffer = Buffer.from(await file.arrayBuffer());

	await client.send(new PutObjectCommand({
		Bucket: R2_BUCKET,
		Key: key,
		Body: buffer,
		ContentType: file.type || "video/mp4",
		ContentLength: buffer.length,
	}));

	return c.json({ ok: true, key, size: buffer.length, publicUrl: R2_PUBLIC_URL ? `${R2_PUBLIC_URL}/${key}` : undefined }, 201);
});

// GET /api/info — module info
routes.get("/api/info", (c) => {
	return c.json({
		module: "tube",
		name: "rTube",
		r2Configured: !!R2_ENDPOINT,
		rtmpIngest: "rtmp://rtube.online:1936/live",
		features: ["video-library", "hls-streaming", "rtmp-ingest"],
	});
});

// GET /api/health
routes.get("/api/health", (c) => c.json({ ok: true }));

// ── 360° Split routes ──

// POST /api/360split — start a split job
routes.post("/api/360split", async (c) => {
	const authToken = extractToken(c.req.raw.headers);
	if (!authToken) return c.json({ error: "Authentication required" }, 401);
	try { await verifyEncryptIDToken(authToken); } catch { return c.json({ error: "Invalid token" }, 401); }

	if (!SPLIT_360_URL) return c.json({ error: "360split service not configured" }, 503);

	const client = getS3();
	if (!client) return c.json({ error: "R2 not configured" }, 503);

	const { videoName, numViews, hFov, vFov, overlap, outputRes } = await c.req.json();
	if (!videoName) return c.json({ error: "videoName required" }, 400);

	try {
		// Fetch video from R2
		const obj = await client.send(new GetObjectCommand({ Bucket: R2_BUCKET, Key: videoName }));
		const bytes = await obj.Body!.transformToByteArray();

		// Build multipart form for 360split
		const form = new FormData();
		form.append("video", new Blob([bytes.buffer]), videoName);
		if (numViews) form.append("num_views", String(numViews));
		if (hFov) form.append("h_fov", String(hFov));
		if (vFov) form.append("v_fov", String(vFov));
		if (overlap) form.append("overlap", String(overlap));
		if (outputRes) form.append("output_res", outputRes);

		const resp = await fetch(`${SPLIT_360_URL}/upload`, { method: "POST", body: form });
		if (!resp.ok) {
			const text = await resp.text();
			return c.json({ error: `360split upload failed: ${text}` }, 502);
		}
		const data = await resp.json();
		return c.json({ jobId: data.job_id });
	} catch (e: any) {
		if (e.name === "NoSuchKey") return c.json({ error: "Video not found in library" }, 404);
		console.error("[Tube] 360split start error:", e);
		return c.json({ error: "Failed to start split job" }, 500);
	}
});

// GET /api/360split/status/:jobId — poll job status
routes.get("/api/360split/status/:jobId", async (c) => {
	if (!SPLIT_360_URL) return c.json({ error: "360split service not configured" }, 503);

	const jobId = c.req.param("jobId");
	try {
		const resp = await fetch(`${SPLIT_360_URL}/status/${jobId}`);
		if (!resp.ok) return c.json({ error: "Status check failed" }, resp.status as any);
		return c.json(await resp.json());
	} catch (e) {
		console.error("[Tube] 360split status error:", e);
		return c.json({ error: "Cannot reach 360split service" }, 502);
	}
});

// POST /api/360split/import/:jobId — import results to R2
routes.post("/api/360split/import/:jobId", async (c) => {
	const authToken = extractToken(c.req.raw.headers);
	if (!authToken) return c.json({ error: "Authentication required" }, 401);
	try { await verifyEncryptIDToken(authToken); } catch { return c.json({ error: "Invalid token" }, 401); }

	if (!SPLIT_360_URL) return c.json({ error: "360split service not configured" }, 503);
	const client = getS3();
	if (!client) return c.json({ error: "R2 not configured" }, 503);

	const jobId = c.req.param("jobId");
	const { originalName } = await c.req.json().catch(() => ({ originalName: "" }));

	try {
		// Get output file list from status
		const statusResp = await fetch(`${SPLIT_360_URL}/status/${jobId}`);
		if (!statusResp.ok) return c.json({ error: "Could not get job status" }, 502);
		const status = await statusResp.json();
		if (status.status !== "complete") return c.json({ error: "Job not complete yet" }, 400);

		const baseName = originalName
			? originalName.replace(/\.[^.]+$/, "")
			: `video-${jobId}`;
		const imported: string[] = [];

		for (const filename of status.output_files || []) {
			// Download from 360split
			const dlResp = await fetch(`${SPLIT_360_URL}/download/${jobId}/${filename}`);
			if (!dlResp.ok) {
				console.error(`[Tube] 360split download failed: ${filename}`);
				continue;
			}
			const buffer = Buffer.from(await dlResp.arrayBuffer());
			const key = `360split/${baseName}/${filename}`;

			await client.send(new PutObjectCommand({
				Bucket: R2_BUCKET,
				Key: key,
				Body: buffer,
				ContentType: "video/mp4",
				ContentLength: buffer.length,
			}));
			imported.push(key);
		}

		// Cleanup temp files on 360split
		await fetch(`${SPLIT_360_URL}/cleanup/${jobId}`, { method: "POST" }).catch(() => {});

		return c.json({ ok: true, imported });
	} catch (e) {
		console.error("[Tube] 360split import error:", e);
		return c.json({ error: "Import failed" }, 500);
	}
});

// ── Live 360° Split routes ──

// POST /api/live-split — start live splitting
routes.post("/api/live-split", async (c) => {
	const authToken = extractToken(c.req.raw.headers);
	if (!authToken) return c.json({ error: "Authentication required" }, 401);
	try { await verifyEncryptIDToken(authToken); } catch { return c.json({ error: "Invalid token" }, 401); }

	if (!SPLIT_360_URL) return c.json({ error: "360split service not configured" }, 503);

	const { streamKey, numViews, hFov, vFov, overlap, outputRes } = await c.req.json();
	if (!streamKey) return c.json({ error: "streamKey required" }, 400);

	const streamUrl = `rtmp://rtube-rtmp:1935/live/${streamKey}`;

	try {
		const resp = await fetch(`${SPLIT_360_URL}/live-split`, {
			method: "POST",
			headers: { "Content-Type": "application/json" },
			body: JSON.stringify({
				stream_url: streamUrl,
				num_views: numViews || 4,
				h_fov: hFov,
				v_fov: vFov,
				overlap: overlap || 0,
				output_res: outputRes || "",
			}),
		});
		if (!resp.ok) {
			const text = await resp.text();
			return c.json({ error: `live-split start failed: ${text}` }, 502);
		}
		return c.json(await resp.json());
	} catch (e: any) {
		console.error("[Tube] live-split start error:", e);
		return c.json({ error: "Cannot reach 360split service" }, 502);
	}
});

// GET /api/live-split/status/:sessionId — proxy status
routes.get("/api/live-split/status/:sessionId", async (c) => {
	if (!SPLIT_360_URL) return c.json({ error: "360split service not configured" }, 503);
	const sessionId = c.req.param("sessionId");
	try {
		const resp = await fetch(`${SPLIT_360_URL}/live-split/status/${sessionId}`);
		if (!resp.ok) return c.json({ error: "Status check failed" }, resp.status as any);
		return c.json(await resp.json());
	} catch (e) {
		console.error("[Tube] live-split status error:", e);
		return c.json({ error: "Cannot reach 360split service" }, 502);
	}
});

// POST /api/live-split/stop/:sessionId — stop session (auth-gated)
routes.post("/api/live-split/stop/:sessionId", async (c) => {
	const authToken = extractToken(c.req.raw.headers);
	if (!authToken) return c.json({ error: "Authentication required" }, 401);
	try { await verifyEncryptIDToken(authToken); } catch { return c.json({ error: "Invalid token" }, 401); }

	if (!SPLIT_360_URL) return c.json({ error: "360split service not configured" }, 503);
	const sessionId = c.req.param("sessionId");
	try {
		const resp = await fetch(`${SPLIT_360_URL}/live-split/stop/${sessionId}`, { method: "POST" });
		if (!resp.ok) return c.json({ error: "Stop failed" }, resp.status as any);
		return c.json(await resp.json());
	} catch (e) {
		console.error("[Tube] live-split stop error:", e);
		return c.json({ error: "Cannot reach 360split service" }, 502);
	}
});

// GET /api/live-split/hls/:sessionId/* — proxy HLS segments
routes.get("/api/live-split/hls/:sessionId/*", async (c) => {
	if (!SPLIT_360_URL) return c.json({ error: "360split service not configured" }, 503);
	const sessionId = c.req.param("sessionId");
	const subpath = c.req.path.replace(new RegExp(`^.*/api/live-split/hls/${sessionId}/`), "");
	if (!subpath) return c.json({ error: "Path required" }, 400);

	try {
		const resp = await fetch(`${SPLIT_360_URL}/live-split/hls/${sessionId}/${subpath}`);
		if (!resp.ok) return new Response("Not found", { status: 404 });

		const contentType = subpath.endsWith(".m3u8")
			? "application/vnd.apple.mpegurl"
			: subpath.endsWith(".ts")
				? "video/mp2t"
				: "application/octet-stream";

		return new Response(resp.body, {
			headers: {
				"Content-Type": contentType,
				"Access-Control-Allow-Origin": "*",
				"Cache-Control": "no-cache, no-store",
			},
		});
	} catch (e) {
		console.error("[Tube] live-split HLS proxy error:", e);
		return new Response("Proxy error", { status: 502 });
	}
});

// ── Page route ──
routes.get("/", (c) => {
	const space = c.req.param("space") || "demo";
	const dataSpace = c.get("effectiveSpace") || space;
	return c.html(renderShell({
		title: `${space} — Tube | rSpace`,
		moduleId: "rtube",
		spaceSlug: space,
		modules: getModuleInfoList(),
		theme: "dark",
		body: `<folk-video-player space="${space}"></folk-video-player>`,
		scripts: `<script type="module" src="/modules/rtube/folk-video-player.js"></script>`,
		styles: `<link rel="stylesheet" href="/modules/rtube/tube.css">`,
	}));
});

export const tubeModule: RSpaceModule = {
	id: "rtube",
	name: "rTube",
	icon: "🎬",
	description: "Community video hosting & live streaming",
	scoping: { defaultScope: 'global', userConfigurable: true },
	routes,
	landingPage: renderLanding,
	standaloneDomain: "rtube.online",
	feeds: [
		{
			id: "videos",
			name: "Videos",
			kind: "data",
			description: "Video library — uploaded and live-streamed content",
			filterable: true,
		},
		{
			id: "watch-activity",
			name: "Watch Activity",
			kind: "attention",
			description: "View counts, watch time, and streaming engagement metrics",
		},
	],
	acceptsFeeds: ["data", "resource"],
	outputPaths: [
		{ path: "videos", name: "Videos", icon: "🎬", description: "Hosted videos and recordings" },
		{ path: "playlists", name: "Playlists", icon: "📺", description: "Video playlists and channels" },
	],
};