mirror of
https://github.com/linsa-io/linsa.git
synced 2026-01-12 12:20:23 +01:00
Add a new "test-jazz-stream" task to flow.toml for testing live stream recording flow, including setup instructions and verification steps.
Update CommentBox component to handle image uploads with validation, preview, and progress tracking, and to manage comments with Jazz container initialization.
This commit is contained in:
Nikita
@@ -0,0 +1,234 @@
|
||||
import { createFileRoute } from "@tanstack/react-router"
|
||||
import { promises as fs } from "fs"
|
||||
|
||||
/**
|
||||
* API endpoint for stream-guard Rust server to upload live stream chunks
|
||||
* Chunks are stored temporarily and then synced to Jazz FileStream by client
|
||||
*/
|
||||
|
||||
const STORAGE_PATH = "/Users/nikiv/fork-i/garden-co/jazz/glide-storage/stream-recordings"
|
||||
|
||||
interface StreamChunk {
|
||||
streamId: string
|
||||
chunkIndex: number
|
||||
data: string // base64 encoded video data
|
||||
timestamp: number
|
||||
metadata?: {
|
||||
width?: number
|
||||
height?: number
|
||||
fps?: number
|
||||
bitrate?: number
|
||||
}
|
||||
}
|
||||
|
||||
interface StreamMetadata {
|
||||
streamId: string
|
||||
title: string
|
||||
startedAt: number
|
||||
streamKey: string
|
||||
metadata?: {
|
||||
width?: number
|
||||
height?: number
|
||||
fps?: number
|
||||
bitrate?: number
|
||||
}
|
||||
}
|
||||
|
||||
async function ensureStorageDir() {
|
||||
await fs.mkdir(STORAGE_PATH, { recursive: true })
|
||||
}
|
||||
|
||||
// POST /api/stream-recording/start - Start a new recording session
|
||||
const startRecording = async ({ request }: { request: Request }) => {
|
||||
try {
|
||||
const body = (await request.json()) as StreamMetadata
|
||||
|
||||
if (!body.streamId || !body.title || !body.streamKey) {
|
||||
return new Response(
|
||||
JSON.stringify({ error: "Missing required fields: streamId, title, streamKey" }),
|
||||
{ status: 400, headers: { "content-type": "application/json" } }
|
||||
)
|
||||
}
|
||||
|
||||
await ensureStorageDir()
|
||||
|
||||
// Create metadata file for this stream
|
||||
const metadataPath = `${STORAGE_PATH}/${body.streamId}-metadata.json`
|
||||
const metadata = {
|
||||
...body,
|
||||
chunks: [],
|
||||
status: "recording",
|
||||
}
|
||||
|
||||
await fs.writeFile(metadataPath, JSON.stringify(metadata, null, 2))
|
||||
|
||||
// Create chunks directory for this stream
|
||||
const chunksDir = `${STORAGE_PATH}/${body.streamId}`
|
||||
await fs.mkdir(chunksDir, { recursive: true })
|
||||
|
||||
console.log(`[stream-recording] Started recording: ${body.streamId}`)
|
||||
|
||||
return new Response(
|
||||
JSON.stringify({ success: true, streamId: body.streamId }),
|
||||
{ status: 200, headers: { "content-type": "application/json" } }
|
||||
)
|
||||
} catch (error) {
|
||||
console.error("[stream-recording] Start error:", error)
|
||||
return new Response(
|
||||
JSON.stringify({ error: "Internal server error" }),
|
||||
{ status: 500, headers: { "content-type": "application/json" } }
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
// POST /api/stream-recording/chunk - Upload a video chunk
|
||||
const uploadChunk = async ({ request }: { request: Request }) => {
|
||||
try {
|
||||
const body = (await request.json()) as StreamChunk
|
||||
|
||||
if (!body.streamId || body.chunkIndex === undefined || !body.data) {
|
||||
return new Response(
|
||||
JSON.stringify({ error: "Missing required fields: streamId, chunkIndex, data" }),
|
||||
{ status: 400, headers: { "content-type": "application/json" } }
|
||||
)
|
||||
}
|
||||
|
||||
await ensureStorageDir()
|
||||
|
||||
// Write chunk to disk
|
||||
const chunkPath = `${STORAGE_PATH}/${body.streamId}/chunk-${String(body.chunkIndex).padStart(6, "0")}.bin`
|
||||
const chunkData = Buffer.from(body.data, "base64")
|
||||
await fs.writeFile(chunkPath, chunkData)
|
||||
|
||||
// Update metadata with chunk info
|
||||
const metadataPath = `${STORAGE_PATH}/${body.streamId}-metadata.json`
|
||||
try {
|
||||
const metadataContent = await fs.readFile(metadataPath, "utf-8")
|
||||
const metadata = JSON.parse(metadataContent)
|
||||
metadata.chunks.push({
|
||||
index: body.chunkIndex,
|
||||
timestamp: body.timestamp,
|
||||
size: chunkData.length,
|
||||
})
|
||||
metadata.lastChunkAt = body.timestamp
|
||||
if (body.metadata) {
|
||||
metadata.metadata = { ...metadata.metadata, ...body.metadata }
|
||||
}
|
||||
await fs.writeFile(metadataPath, JSON.stringify(metadata, null, 2))
|
||||
} catch (err) {
|
||||
console.warn(`[stream-recording] Could not update metadata for ${body.streamId}:`, err)
|
||||
}
|
||||
|
||||
return new Response(
|
||||
JSON.stringify({ success: true, chunkIndex: body.chunkIndex }),
|
||||
{ status: 200, headers: { "content-type": "application/json" } }
|
||||
)
|
||||
} catch (error) {
|
||||
console.error("[stream-recording] Chunk upload error:", error)
|
||||
return new Response(
|
||||
JSON.stringify({ error: "Internal server error" }),
|
||||
{ status: 500, headers: { "content-type": "application/json" } }
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
// POST /api/stream-recording/end - End a recording session
|
||||
const endRecording = async ({ request }: { request: Request }) => {
|
||||
try {
|
||||
const body = (await request.json()) as { streamId: string; endedAt: number }
|
||||
|
||||
if (!body.streamId || !body.endedAt) {
|
||||
return new Response(
|
||||
JSON.stringify({ error: "Missing required fields: streamId, endedAt" }),
|
||||
{ status: 400, headers: { "content-type": "application/json" } }
|
||||
)
|
||||
}
|
||||
|
||||
// Update metadata to mark as ended
|
||||
const metadataPath = `${STORAGE_PATH}/${body.streamId}-metadata.json`
|
||||
try {
|
||||
const metadataContent = await fs.readFile(metadataPath, "utf-8")
|
||||
const metadata = JSON.parse(metadataContent)
|
||||
metadata.endedAt = body.endedAt
|
||||
metadata.status = "ended"
|
||||
metadata.durationMs = body.endedAt - metadata.startedAt
|
||||
await fs.writeFile(metadataPath, JSON.stringify(metadata, null, 2))
|
||||
|
||||
console.log(`[stream-recording] Ended recording: ${body.streamId}`)
|
||||
|
||||
return new Response(
|
||||
JSON.stringify({ success: true, streamId: body.streamId }),
|
||||
{ status: 200, headers: { "content-type": "application/json" } }
|
||||
)
|
||||
} catch (err) {
|
||||
return new Response(
|
||||
JSON.stringify({ error: "Stream not found" }),
|
||||
{ status: 404, headers: { "content-type": "application/json" } }
|
||||
)
|
||||
}
|
||||
} catch (error) {
|
||||
console.error("[stream-recording] End error:", error)
|
||||
return new Response(
|
||||
JSON.stringify({ error: "Internal server error" }),
|
||||
{ status: 500, headers: { "content-type": "application/json" } }
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
// GET /api/stream-recording/list - List all recordings
|
||||
const listRecordings = async ({ request }: { request: Request }) => {
|
||||
try {
|
||||
await ensureStorageDir()
|
||||
|
||||
const files = await fs.readdir(STORAGE_PATH)
|
||||
const metadataFiles = files.filter((f) => f.endsWith("-metadata.json"))
|
||||
|
||||
const recordings = []
|
||||
for (const file of metadataFiles) {
|
||||
try {
|
||||
const content = await fs.readFile(`${STORAGE_PATH}/${file}`, "utf-8")
|
||||
const metadata = JSON.parse(content)
|
||||
recordings.push(metadata)
|
||||
} catch (err) {
|
||||
console.warn(`[stream-recording] Could not read ${file}:`, err)
|
||||
}
|
||||
}
|
||||
|
||||
return new Response(
|
||||
JSON.stringify({ recordings }),
|
||||
{ status: 200, headers: { "content-type": "application/json" } }
|
||||
)
|
||||
} catch (error) {
|
||||
console.error("[stream-recording] List error:", error)
|
||||
return new Response(
|
||||
JSON.stringify({ error: "Internal server error" }),
|
||||
{ status: 500, headers: { "content-type": "application/json" } }
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
export const Route = createFileRoute("/api/stream-recording")({
|
||||
server: {
|
||||
handlers: {
|
||||
GET: listRecordings,
|
||||
POST: (ctx) => {
|
||||
const url = new URL(ctx.request.url)
|
||||
const action = url.searchParams.get("action")
|
||||
|
||||
switch (action) {
|
||||
case "start":
|
||||
return startRecording(ctx)
|
||||
case "chunk":
|
||||
return uploadChunk(ctx)
|
||||
case "end":
|
||||
return endRecording(ctx)
|
||||
default:
|
||||
return new Response(
|
||||
JSON.stringify({ error: "Invalid action. Use ?action=start|chunk|end" }),
|
||||
{ status: 400, headers: { "content-type": "application/json" } }
|
||||
)
|
||||
}
|
||||
},
|
||||
},
|
||||
},
|
||||
})
|
||||
@@ -0,0 +1,212 @@
|
||||
import { useState, useEffect } from "react"
|
||||
import { createFileRoute } from "@tanstack/react-router"
|
||||
import { useAccount } from "jazz-tools/react"
|
||||
import { ViewerAccount, type StreamRecording, StreamRecordingList } from "@/lib/jazz/schema"
|
||||
import { StreamTimeline } from "@/components/StreamTimeline"
|
||||
import { Video, RefreshCw } from "lucide-react"
|
||||
import { co } from "jazz-tools"
|
||||
|
||||
export const Route = createFileRoute("/streams")({
|
||||
component: StreamsPage,
|
||||
ssr: false,
|
||||
})
|
||||
|
||||
function StreamsPage() {
|
||||
const me = useAccount(ViewerAccount)
|
||||
const [syncing, setSyncing] = useState(false)
|
||||
const [lastSync, setLastSync] = useState<Date | null>(null)
|
||||
|
||||
const root = me.$isLoaded ? me.root : null
|
||||
const recordingsList = root?.$isLoaded ? root.streamRecordings : null
|
||||
|
||||
// Auto-sync pending recordings from API every 5 seconds
|
||||
useEffect(() => {
|
||||
const interval = setInterval(() => {
|
||||
void syncPendingRecordings()
|
||||
}, 5000)
|
||||
return () => clearInterval(interval)
|
||||
}, [root])
|
||||
|
||||
const syncPendingRecordings = async () => {
|
||||
if (!root?.streamRecordings?.$isLoaded || syncing) return
|
||||
|
||||
setSyncing(true)
|
||||
try {
|
||||
// Fetch pending recordings from stream-guard API
|
||||
const response = await fetch("/api/stream-recording")
|
||||
if (!response.ok) {
|
||||
console.error("[streams] Failed to fetch recordings")
|
||||
return
|
||||
}
|
||||
|
||||
const data = (await response.json()) as {
|
||||
recordings: Array<{
|
||||
streamId: string
|
||||
title: string
|
||||
startedAt: number
|
||||
endedAt?: number
|
||||
streamKey: string
|
||||
status: string
|
||||
chunks: Array<{ index: number; timestamp: number; size: number }>
|
||||
metadata?: {
|
||||
width?: number
|
||||
height?: number
|
||||
fps?: number
|
||||
bitrate?: number
|
||||
}
|
||||
}>
|
||||
}
|
||||
|
||||
const pendingRecordings = data.recordings
|
||||
|
||||
if (pendingRecordings.length === 0) {
|
||||
return
|
||||
}
|
||||
|
||||
console.log(`[streams] Found ${pendingRecordings.length} recordings to sync`)
|
||||
|
||||
// Get existing IDs to avoid duplicates
|
||||
const existingKeys = new Set(
|
||||
root.streamRecordings.$isLoaded
|
||||
? [...root.streamRecordings].map((item) => item.streamKey)
|
||||
: []
|
||||
)
|
||||
|
||||
// Process each recording
|
||||
for (const rec of pendingRecordings) {
|
||||
if (existingKeys.has(rec.streamKey)) {
|
||||
// Update existing recording
|
||||
const existing = [...root.streamRecordings].find(
|
||||
(r) => r.streamKey === rec.streamKey
|
||||
)
|
||||
if (existing && rec.endedAt && !existing.endedAt) {
|
||||
// Mark as ended
|
||||
existing.endedAt = rec.endedAt
|
||||
existing.isLive = false
|
||||
existing.durationMs = rec.endedAt - rec.startedAt
|
||||
}
|
||||
continue
|
||||
}
|
||||
|
||||
// Create new recording in Jazz
|
||||
try {
|
||||
// Create FileStream from chunks
|
||||
const fileStream = co.fileStream().create({ owner: me })
|
||||
|
||||
// Start the stream with metadata
|
||||
fileStream.start({
|
||||
mimeType: "video/x-matroska", // .mkv format
|
||||
fileName: `${rec.title}.mkv`,
|
||||
totalSizeBytes: rec.chunks.reduce((sum, c) => sum + c.size, 0),
|
||||
})
|
||||
|
||||
// Fetch and push chunks
|
||||
for (const chunk of rec.chunks) {
|
||||
try {
|
||||
const chunkPath = `/Users/nikiv/fork-i/garden-co/jazz/glide-storage/stream-recordings/${rec.streamId}/chunk-${String(chunk.index).padStart(6, "0")}.bin`
|
||||
const chunkData = await fetch(
|
||||
`/api/stream-recording/chunk?path=${encodeURIComponent(chunkPath)}`
|
||||
).then((r) => r.arrayBuffer())
|
||||
|
||||
fileStream.push(new Uint8Array(chunkData))
|
||||
} catch (err) {
|
||||
console.error(`[streams] Failed to fetch chunk ${chunk.index}:`, err)
|
||||
}
|
||||
}
|
||||
|
||||
// End the stream if recording is complete
|
||||
if (rec.status === "ended") {
|
||||
fileStream.end()
|
||||
}
|
||||
|
||||
// Create StreamRecording object
|
||||
const recording = {
|
||||
title: rec.title,
|
||||
startedAt: rec.startedAt,
|
||||
endedAt: rec.endedAt || null,
|
||||
durationMs: rec.endedAt
|
||||
? rec.endedAt - rec.startedAt
|
||||
: Date.now() - rec.startedAt,
|
||||
streamKey: rec.streamKey,
|
||||
isLive: rec.status === "recording",
|
||||
videoFile: fileStream,
|
||||
thumbnailData: null,
|
||||
metadata: rec.metadata || null,
|
||||
}
|
||||
|
||||
// Push to Jazz
|
||||
root.streamRecordings.$jazz.push(recording)
|
||||
console.log(`[streams] Added recording to Jazz: ${rec.title}`)
|
||||
setLastSync(new Date())
|
||||
} catch (err) {
|
||||
console.error(`[streams] Failed to create recording in Jazz:`, err)
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
console.error("[streams] Sync error:", error)
|
||||
} finally {
|
||||
setSyncing(false)
|
||||
}
|
||||
}
|
||||
|
||||
const handleManualSync = () => {
|
||||
void syncPendingRecordings()
|
||||
}
|
||||
|
||||
if (!me.$isLoaded || !root?.$isLoaded) {
|
||||
return (
|
||||
<div className="min-h-screen text-white grid place-items-center">
|
||||
<p className="text-slate-400">Loading Jazz...</p>
|
||||
</div>
|
||||
)
|
||||
}
|
||||
|
||||
const recordings: StreamRecording[] = recordingsList?.$isLoaded
|
||||
? [...recordingsList]
|
||||
: []
|
||||
|
||||
return (
|
||||
<div className="min-h-screen text-white">
|
||||
<div className="max-w-6xl mx-auto px-4 py-10">
|
||||
<div className="flex items-center justify-between mb-8">
|
||||
<div className="flex items-center gap-3">
|
||||
<Video className="w-6 h-6 text-teal-400" />
|
||||
<h1 className="text-2xl font-semibold">Live Stream Recordings</h1>
|
||||
</div>
|
||||
<div className="flex items-center gap-3">
|
||||
{lastSync && (
|
||||
<span className="text-xs text-slate-400">
|
||||
Last sync: {lastSync.toLocaleTimeString()}
|
||||
</span>
|
||||
)}
|
||||
<button
|
||||
type="button"
|
||||
onClick={handleManualSync}
|
||||
disabled={syncing}
|
||||
className="inline-flex items-center gap-2 px-4 py-2 rounded-lg text-sm font-medium text-white bg-teal-600 hover:bg-teal-500 transition-colors disabled:opacity-50 disabled:cursor-not-allowed"
|
||||
>
|
||||
<RefreshCw className={`w-4 h-4 ${syncing ? "animate-spin" : ""}`} />
|
||||
{syncing ? "Syncing..." : "Sync Now"}
|
||||
</button>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
{recordings.length === 0 ? (
|
||||
<div className="text-center py-12 text-slate-400">
|
||||
<Video className="w-12 h-12 mx-auto mb-4 opacity-50" />
|
||||
<p>No stream recordings yet</p>
|
||||
<p className="text-sm mt-1">
|
||||
Start streaming to stream-guard to see recordings here
|
||||
</p>
|
||||
</div>
|
||||
) : (
|
||||
<div className="space-y-6">
|
||||
{recordings.map((recording, index) => (
|
||||
<StreamTimeline key={index} recording={recording} width={900} height={120} />
|
||||
))}
|
||||
</div>
|
||||
)}
|
||||
</div>
|
||||
</div>
|
||||
)
|
||||
}
|
||||
Reference in New Issue
Block a user