diff --git a/js/app/src/screens/upload.tsx b/js/app/src/screens/upload.tsx index 7d43126a..655e5c46 100644 --- a/js/app/src/screens/upload.tsx +++ b/js/app/src/screens/upload.tsx @@ -1,52 +1,291 @@ -import { Text, zero } from "@streamplace/components"; +import { + Admonition, + Button, + Checkbox, + MenuContainer, + MenuGroup, + MenuLabel, + MenuSeparator, + Select, + Text, + Tooltip, + useTheme, + View, + zero, +} from "@streamplace/components"; +import { + CONTENT_WARNINGS, + LICENSE_OPTIONS, +} from "@streamplace/components/src/lib/metadata-constants"; import { usePDSAgent } from "@streamplace/components/src/streamplace-store/xrpc"; -import { useCallback, useRef, useState } from "react"; -import { Pressable, ScrollView, View } from "react-native"; +import ActivityPicker from "components/activity-picker"; +import { Image } from "expo-image"; +import { + AlertCircle, + ArrowUp, + CheckCircle2, + ImagePlus, + LoaderCircle, + Video, + X, +} from "lucide-react-native"; +import { useCallback, useEffect, useRef, useState } from "react"; +import { + Animated, + Pressable, + ScrollView, + TextInput, + useWindowDimensions, +} from "react-native"; +import type { PlaceStreamLivestream, PlaceStreamVideo } from "streamplace"; import * as tus from "tus-js-client"; -type Status = +// ── types ──────────────────────────────────────────────────────────────────── + +type TrackRef = { uri: string; cid: string }; + +type UploadPhase = | { kind: "idle" } | { kind: "creating" } | { kind: "uploading"; pct: number; bytesSent: number; bytesTotal: number } - | { kind: "done"; uploadUrl: string } + | { + kind: "processing"; + uploadId: string; + serverStatus?: "pending" | "processing"; + progress?: number; + } + | { kind: "ready"; uploadId: string; tracks: TrackRef[]; durationMs: number } + | { kind: "publishing" } + | { kind: "done"; videoUri: string } | { kind: "error"; message: string }; +// ── helpers ─────────────────────────────────────────────────────────────────── + +function humanBytes(n: number): string { + if (n < 1024) return `${n} B`; + if (n < 1024 * 1024) return `${(n / 1024).toFixed(1)} KB`; + if (n < 1024 * 1024 * 1024) return `${(n / 1024 / 1024).toFixed(1)} MB`; + return `${(n / 1024 / 1024 / 1024).toFixed(2)} GB`; +} + +// Returns an error string if the file doesn't look like a video, null if ok. +// Reads the first 12 bytes and matches magic signatures for common containers. +async function validateVideoFile(file: File): Promise { + const buf = await file.slice(0, 12).arrayBuffer(); + const b = new Uint8Array(buf); + + const eq = (offset: number, ...bytes: number[]) => + bytes.every((v, i) => b[offset + i] === v); + const str = (offset: number, len: number) => + String.fromCharCode(...Array.from(b.slice(offset, offset + len))); + + // MP4 / MOV / M4V — "ftyp" box at offset 4 + if (str(4, 4) === "ftyp") return null; + // WebM / MKV — EBML magic + if (eq(0, 0x1a, 0x45, 0xdf, 0xa3)) return null; + // AVI — "RIFF" header with "AVI " chunk + if (str(0, 4) === "RIFF" && str(8, 4) === "AVI ") return null; + // OGG (video: OGV, Theora) + if (str(0, 4) === "OggS") return null; + // FLV + if (str(0, 3) === "FLV") return null; + // MPEG-TS — 188-byte packets starting with sync byte 0x47 + if (b[0] === 0x47) return null; + + return "File doesn't appear to be a supported video format (MP4, WebM, MKV, MOV, AVI, OGG, FLV, MPEG-TS)."; +} + +const POLL_INTERVAL_MS = 3000; + +// ── screen ─────────────────────────────────────────────────────────────────── + export default function UploadScreen() { const agent = usePDSAgent(); + const { theme, zero: zt } = useTheme(); + const { width } = useWindowDimensions(); + const isWide = width > 800; + + // file const fileInputRef = useRef(null); const uploadRef = useRef(null); + const pollRef = useRef | null>(null); + const processingAnim = useRef(new Animated.Value(0)).current; const [file, setFile] = useState(null); - const [status, setStatus] = useState({ kind: "idle" }); - const pick = useCallback(() => { - fileInputRef.current?.click(); + // upload state machine + const [phase, setPhase] = useState({ kind: "idle" }); + + // indeterminate progress bar animation + useEffect(() => { + if ( + phase.kind === "creating" || + phase.kind === "processing" || + phase.kind === "publishing" + ) { + const anim = Animated.loop( + Animated.sequence([ + Animated.timing(processingAnim, { + toValue: 1, + duration: 1200, + useNativeDriver: false, + }), + Animated.timing(processingAnim, { + toValue: 0, + duration: 1200, + useNativeDriver: false, + }), + ]), + ); + anim.start(); + return () => anim.stop(); + } else { + processingAnim.setValue(0); + } + }, [phase.kind, processingAnim]); + + // metadata form — always editable + const [title, setTitle] = useState(""); + const [description, setDescription] = useState(""); + const [activity, setActivity] = + useState(undefined); + const [tags, setTags] = useState([]); + const [tagInput, setTagInput] = useState(""); + const [thumbnail, setThumbnail] = useState(undefined); + const [thumbnailUrl, setThumbnailUrl] = useState( + undefined, + ); + const thumbnailInputRef = useRef(null); + const [warnings, setWarnings] = useState>(new Set()); + const [license, setLicense] = useState( + "place.stream.metadata.contentRights#all-rights-reserved", + ); + + // cleanup on unmount + useEffect(() => { + return () => { + if (pollRef.current) clearTimeout(pollRef.current); + uploadRef.current?.abort(); + }; }, []); - const handleFileInputChange = useCallback( - (event: React.ChangeEvent) => { - const f = event.target.files?.[0] ?? null; + // clean up old blob URLs when thumbnail changes or unmounts + useEffect(() => { + return () => { + if (thumbnailUrl) URL.revokeObjectURL(thumbnailUrl); + }; + }, [thumbnailUrl]); + + const pickFile = useCallback(() => fileInputRef.current?.click(), []); + + const handleFileChange = useCallback( + (e: React.ChangeEvent) => { + const f = e.target.files?.[0] ?? null; setFile(f); - setStatus({ kind: "idle" }); - event.target.value = ""; + if (f) setTitle((t) => t || f.name.replace(/\.[^.]+$/, "")); + setPhase({ kind: "idle" }); + e.target.value = ""; }, [], ); - const start = useCallback(async () => { + const toggleWarning = useCallback((value: string) => { + setWarnings((prev) => { + const next = new Set(prev); + next.has(value) ? next.delete(value) : next.add(value); + return next; + }); + }, []); + + // thumbnail handlers + const handleThumbnailSelect = useCallback(() => { + thumbnailInputRef.current?.click(); + }, []); + + const handleThumbnailChange = useCallback( + (e: React.ChangeEvent) => { + const f = e.target.files?.[0] ?? null; + if (f) { + const blob = new Blob([f], { type: f.type }); + setThumbnail(blob); + setThumbnailUrl(URL.createObjectURL(blob)); + } + e.target.value = ""; + }, + [], + ); + + const handleThumbnailRemove = useCallback(() => { + setThumbnail(undefined); + setThumbnailUrl(undefined); + }, []); + + // ── polling ────────────────────────────────────────────────────────────── + + const pollStatus = useCallback( + (uploadId: string) => { + if (!agent) return; + const check = async () => { + try { + const res = await agent.place.stream.media.getUploadStatus({ + uploadId, + }); + const data = res.data; + + if (data.status === "done" && data.tracks) { + setPhase({ + kind: "ready", + uploadId, + tracks: data.tracks, + durationMs: data.durationMs ?? 0, + }); + return; + } + if (data.status === "error") { + setPhase({ + kind: "error", + message: data.error ?? "Processing failed", + }); + return; + } + setPhase({ + kind: "processing", + uploadId, + serverStatus: data.status as "pending" | "processing", + progress: data.progress, + }); + pollRef.current = setTimeout(check, POLL_INTERVAL_MS); + } catch { + pollRef.current = setTimeout(check, POLL_INTERVAL_MS); + } + }; + check(); + }, + [agent], + ); + + // ── upload ─────────────────────────────────────────────────────────────── + + const startUpload = useCallback(async () => { if (!agent || !file) return; if (!agent.did) { - setStatus({ kind: "error", message: "Not logged in" }); + setPhase({ kind: "error", message: "Not logged in" }); + return; + } + const validationError = await validateVideoFile(file); + if (validationError) { + setPhase({ kind: "error", message: validationError }); return; } - setStatus({ kind: "creating" }); + const mimeType = file.type.startsWith("video/") ? file.type : "video/mp4"; + setPhase({ kind: "creating" }); try { const res = await agent.place.stream.media.createUpload({ size: file.size, - mimeType: file.type || "application/octet-stream", + mimeType, filename: file.name, }); if (!res.success) throw new Error("createUpload failed"); - const { uploadUrl, uploadToken } = res.data; + const { uploadUrl, uploadToken, uploadId } = res.data; await new Promise((resolve, reject) => { const upload = new tus.Upload(file, { @@ -54,138 +293,1108 @@ export default function UploadScreen() { retryDelays: [0, 1000, 3000, 5000], headers: { Authorization: `Bearer ${uploadToken}` }, metadata: { filename: file.name, filetype: file.type }, - onError(err) { - reject(err); - }, + onError: reject, onProgress(bytesSent, bytesTotal) { - setStatus({ + setPhase({ kind: "uploading", pct: bytesTotal > 0 ? (bytesSent / bytesTotal) * 100 : 0, bytesSent, bytesTotal, }); }, - onSuccess() { - resolve(); - }, + onSuccess: () => resolve(), }); uploadRef.current = upload; upload.start(); }); - setStatus({ kind: "done", uploadUrl }); + uploadRef.current = null; + setPhase({ kind: "processing", uploadId }); + pollStatus(uploadId); } catch (err) { - console.error("upload failed", err); - setStatus({ + uploadRef.current = null; + setPhase({ kind: "error", message: err instanceof Error ? err.message : String(err), }); - } finally { - uploadRef.current = null; } - }, [agent, file]); + }, [agent, file, pollStatus]); - const cancel = useCallback(() => { + const cancelUpload = useCallback(() => { + if (pollRef.current) clearTimeout(pollRef.current); uploadRef.current?.abort(); uploadRef.current = null; - setStatus({ kind: "idle" }); + setPhase({ kind: "idle" }); + }, []); + + // ── publish ────────────────────────────────────────────────────────────── + + const publish = useCallback(async () => { + if (phase.kind !== "ready" || !agent || !agent.did) return; + setPhase({ kind: "publishing" }); + try { + const { tracks, durationMs } = phase; + + const record: PlaceStreamVideo.Record = { + $type: "place.stream.video", + title: title.trim() || file?.name || "Untitled", + durationMs, + source: { + $type: "place.stream.media.defs#sourceTracks", + tracks: tracks.map((t) => ({ + $type: "com.atproto.repo.strongRef", + uri: t.uri, + cid: t.cid, + })), + }, + }; + + if (description.trim()) record.description = description.trim(); + if (activity) record.activity = activity; + if (tags.length > 0) record.tags = tags; + + if (warnings.size > 0) { + const cw: Record = {}; + for (const w of warnings) { + const key = w.split("#")[1]; + if (key) cw[key] = true; + } + record.contentWarnings = { + $type: "place.stream.metadata.contentWarnings", + ...cw, + } as any; + } + + if ( + license && + license !== "place.stream.metadata.contentRights#all-rights-reserved" + ) { + const licenseKey = license.split("#")[1]; + record.contentRights = { + $type: "place.stream.metadata.contentRights", + license: { $type: license } as any, + [licenseKey]: { $type: license } as any, + } as any; + } + + if (thumbnail) { + try { + const blobRes = await agent.uploadBlob(thumbnail, { + encoding: thumbnail.type || "image/jpeg", + }); + if (blobRes.success) record.thumb = blobRes.data.blob as any; + } catch { + // thumbnail failure is non-fatal + } + } + + const rkey = `${Date.now()}`; + const putRes = await agent.com.atproto.repo.putRecord({ + repo: agent.did, + collection: "place.stream.video", + rkey, + record, + }); + + if (!putRes.success) throw new Error("putRecord failed"); + setPhase({ kind: "done", videoUri: putRes.data.uri }); + } catch (err) { + setPhase({ + kind: "error", + message: err instanceof Error ? err.message : String(err), + }); + } + }, [ + phase, + agent, + title, + description, + activity, + tags, + thumbnail, + warnings, + license, + file, + ]); + + // ── derived state ───────────────────────────────────────────────────────── + + const isUploading = + phase.kind === "creating" || + phase.kind === "uploading" || + phase.kind === "processing"; + const isPublishing = phase.kind === "publishing"; + const canUpload = !!file && (phase.kind === "idle" || phase.kind === "error"); + const canPublish = phase.kind === "ready" && !!title.trim(); + + // ── VOD manager ──────────────────────────────────────────────────────────── + + const [managerMode, setManagerMode] = useState<"upload" | "videos">("upload"); + const [userVideos, setUserVideos] = useState([]); + const [editingVideoUri, setEditingVideoUri] = useState( + undefined, + ); + const [updating, setUpdating] = useState(false); + const [deleting, setDeleting] = useState(false); + + const fetchVideos = useCallback(async () => { + if (!agent || !agent.did) return; + try { + const res = await agent.place.stream.media.getVideoList({ + repo: agent.did, + }); + setUserVideos(res.data.videos || []); + } catch (err) { + console.error("Failed to fetch videos", err); + } + }, [agent]); + + useEffect(() => { + if (managerMode === "videos") { + fetchVideos(); + } + }, [managerMode, fetchVideos]); + + const handleSelectVideo = useCallback((video: any) => { + const rec = video.record?.value || video.record || {}; + setEditingVideoUri(video.uri); + setTitle(rec.title || ""); + setDescription(rec.description || ""); + setActivity(rec.activity || undefined); + setTags(rec.tags || []); + setThumbnail(undefined); + setThumbnailUrl(undefined); + // Clear content warnings + const cw = rec.contentWarnings?.warnings || []; + setWarnings(new Set(cw)); + // Set license + const rights = rec.contentRights || {}; + setLicense( + rights.license?.$type || + "place.stream.metadata.contentRights#all-rights-reserved", + ); + setManagerMode("upload"); }, []); + const handleUpdateVideo = useCallback(async () => { + if (!agent || !agent.did || !editingVideoUri) return; + setUpdating(true); + try { + const record: Record = { + $type: "place.stream.video", + title: title.trim(), + source: {}, // placeholder, will be merged with existing + durationMs: 0, // placeholder + }; + // Fetch existing record to preserve source/duration + const existing = await agent.com.atproto.repo.getRecord({ + repo: agent.did, + collection: "place.stream.video", + rkey: editingVideoUri.split("/").pop()!, + }); + const existingRec = existing.data.value as any; + record.source = existingRec.source; + record.durationMs = existingRec.durationMs; + if (description.trim()) record.description = description.trim(); + if (activity) record.activity = activity; + if (tags.length > 0) record.tags = tags; + if (warnings.size > 0) { + const cw: Record = {}; + for (const w of warnings) { + const key = w.split("#")[1]; + if (key) cw[key] = true; + } + record.contentWarnings = { + $type: "place.stream.metadata.contentWarnings", + ...cw, + }; + } + if ( + license && + license !== "place.stream.metadata.contentRights#all-rights-reserved" + ) { + const licenseKey = license.split("#")[1]; + record.contentRights = { + $type: "place.stream.metadata.contentRights", + license: { $type: license }, + [licenseKey]: { $type: license }, + }; + } + if (thumbnail) { + try { + const blobRes = await agent.uploadBlob(thumbnail, { + encoding: thumbnail.type || "image/jpeg", + }); + if (blobRes.success) record.thumb = blobRes.data.blob; + } catch { + // thumbnail is non-fatal + } + } + await agent.com.atproto.repo.putRecord({ + repo: agent.did, + collection: "place.stream.video", + rkey: editingVideoUri.split("/").pop()!, + record: record as any, + }); + setEditingVideoUri(undefined); + setFile(null); + setPhase({ kind: "idle" }); + } catch (err) { + console.error("Failed to update video", err); + } finally { + setUpdating(false); + } + }, [ + agent, + editingVideoUri, + title, + description, + activity, + tags, + warnings, + license, + thumbnail, + ]); + + const handleDeleteVideo = useCallback(async () => { + if (!agent || !editingVideoUri) return; + setDeleting(true); + try { + await agent.com.atproto.repo.deleteRecord({ + repo: agent.did!, + collection: "place.stream.video", + rkey: editingVideoUri.split("/").pop()!, + }); + setEditingVideoUri(undefined); + setFile(null); + setPhase({ kind: "idle" }); + setManagerMode("videos"); + } catch (err) { + console.error("Failed to delete video", err); + } finally { + setDeleting(false); + } + }, [agent, editingVideoUri]); + + // ── render ──────────────────────────────────────────────────────────────── + return ( - - Resumable upload (TUS) - - Developer test page for /api/upload. Pick a file, hit upload, watch - bytes go. - - - - {file ? `Selected: ${file.name}` : "Choose a file"} - {file && ( - - {file.type || "unknown"} — {humanBytes(file.size)} - - )} - - - {file && status.kind !== "uploading" && ( - + + {/* Mode toggle */} + - - {status.kind === "creating" ? "Creating upload…" : "Start upload"} - - - )} - - {status.kind === "uploading" && ( - - - Uploading: {status.pct.toFixed(1)}% ({humanBytes(status.bytesSent)}{" "} - / {humanBytes(status.bytesTotal)}) - - setManagerMode("upload")} + style={[ + zero.px[4], + zero.py[2], + zero.r.lg, + { + borderBottomRightRadius: 0, + borderTopRightRadius: 0, + }, + managerMode === "upload" + ? { backgroundColor: theme.colors.primary } + : { + backgroundColor: "transparent", + borderWidth: 1, + borderColor: theme.colors.border, + }, + ]} > - - + + Upload + + setManagerMode("videos")} style={[ - zero.p[3], - { backgroundColor: "#722", borderRadius: 8 }, - zero.layout.flex.center, + zero.px[4], + zero.py[2], + zero.r.lg, + { + borderBottomLeftRadius: 0, + borderTopLeftRadius: 0, + }, + managerMode === "videos" + ? { backgroundColor: theme.colors.primary } + : { + backgroundColor: "transparent", + borderWidth: 1, + borderColor: theme.colors.border, + borderLeftWidth: 0, + }, ]} > - Cancel + + My Videos + - )} - {status.kind === "done" && ( - - Upload complete: {status.uploadUrl} - - )} + {/* Video list mode */} + {managerMode === "videos" && ( + + {userVideos.length === 0 && ( + + + )} + {userVideos.map((video: any) => { + const rec = video.record?.value || video.record || {}; + const thumb = rec.thumb; + const thumbUrl = thumb + ? `https://cdn.stream.place/thumb/${thumb.ref?.$link || thumb.cid || ""}` + : undefined; + return ( + handleSelectVideo(video)} + style={[ + zero.p[3], + zero.r.md, + { + flexDirection: "row", + gap: 12, + borderWidth: 1, + borderColor: theme.colors.border, + backgroundColor: theme.colors.background, + alignItems: "center", + }, + ]} + > + {thumbUrl ? ( + + ) : ( + + + )} + + + {rec.title || "Untitled"} + + + {video.viewCounts?.count != null && + `${video.viewCounts.count} views · `} + {rec.durationMs + ? `${Math.round(rec.durationMs / 1000)}s` + : ""} + + + + ); + })} + + )} + + {/* Upload/Edit mode */} + {managerMode === "upload" && ( + <> + {editingVideoUri && ( + + + Editing video + + + + { + setEditingVideoUri(undefined); + setPhase({ kind: "idle" }); + setFile(null); + setTitle(""); + setDescription(""); + setActivity(undefined); + setTags([]); + setTagInput(""); + setThumbnail(undefined); + setThumbnailUrl(undefined); + setWarnings(new Set()); + setLicense( + "place.stream.metadata.contentRights#all-rights-reserved", + ); + }} + > + + + + + )} + + {/* ── left column: metadata ────────────────────────────────────── */} + + + {/* title + description */} + + Details + + + + Title + + + + + + + Description + + + + + + + {/* activity */} + + Activity + + + + + + + + {/* tags */} + + Tags + + + {tags.length > 0 && ( + + {tags.map((tag) => ( + + setTags(tags.filter((t) => t !== tag)) + } + style={{ + flexDirection: "row", + alignItems: "center", + backgroundColor: theme.colors.primary + "22", + borderRadius: 12, + paddingHorizontal: 10, + paddingVertical: 3, + }} + > + + {tag} + + + × + + + ))} + + )} + {tags.length < 10 && ( + + setTagInput(v.replace(/[^a-zA-Z0-9:]/g, "")) + } + placeholder="Add tag, press Enter" + placeholderTextColor={theme.colors.mutedForeground} + returnKeyType="done" + onSubmitEditing={() => { + const t = tagInput.trim(); + if (t && !tags.includes(t)) setTags([...tags, t]); + setTagInput(""); + }} + style={inputStyle(theme)} + /> + )} + + + + + {/* content warnings */} + + Content Warnings + + + {CONTENT_WARNINGS.map((cw) => ( + + toggleWarning(cw.value)} + label={cw.label} + size="sm" + /> + + ))} + + + - {status.kind === "error" && ( - Error: {status.message} - )} + {/* license */} + + License + + + + + + + {/* status */} + {phase.kind !== "idle" && ( + + Status + + {phase.kind === "creating" && ( + + + + + Preparing upload… + + + + + + + )} + {phase.kind === "uploading" && ( + + + + + {phase.pct.toFixed(1)}% —{" "} + {humanBytes(phase.bytesSent)} /{" "} + {humanBytes(phase.bytesTotal)} + + + + + + + )} + {(phase.kind === "processing" || + phase.kind === "publishing") && ( + + + + + {phase.kind === "processing" + ? phase.serverStatus === "processing" + ? `Processing video${phase.progress != null ? ` (${phase.progress}%)` : "…"}` + : "Waiting to process…" + : "Publishing…"} + + + {phase.kind === "processing" && + phase.progress != null ? ( + + + + ) : ( + + + + )} + + )} + {phase.kind === "ready" && ( + + + + Ready to publish + + + )} + {phase.kind === "done" && ( + + + + Published + + + )} + {phase.kind === "error" && ( + + + + {phase.message} + + + )} + + + )} + + {/* actions */} + + + {editingVideoUri ? ( + <> + + + ) : ( + <> + {canUpload && ( + + )} + {!file && phase.kind === "idle" && ( + + )} + {isUploading && phase.kind !== "processing" && ( + + )} + {(phase.kind === "ready" || isPublishing) && ( + + )} + + )} + + + + + + )} + + + )} + ); } -function humanBytes(n: number): string { - if (n < 1024) return `${n} B`; - if (n < 1024 * 1024) return `${(n / 1024).toFixed(1)} KB`; - if (n < 1024 * 1024 * 1024) return `${(n / 1024 / 1024).toFixed(1)} MB`; - return `${(n / 1024 / 1024 / 1024).toFixed(2)} GB`; +// ── style helpers ───────────────────────────────────────────────────────────── + +function inputStyle(theme: any) { + return { + borderWidth: 1, + borderColor: theme.colors.border, + borderRadius: 8, + padding: 8, + color: theme.colors.foreground, + backgroundColor: theme.colors.background, + }; } diff --git a/js/components/src/components/ui/index.ts b/js/components/src/components/ui/index.ts index 95d66abe..b42c3639 100644 --- a/js/components/src/components/ui/index.ts +++ b/js/components/src/components/ui/index.ts @@ -18,6 +18,7 @@ export * from "./loader"; export * from "./menu"; export * from "./portal"; export * from "./resizeable"; +export * from "./select"; export * from "./slider"; export * from "./text"; export * from "./textarea"; diff --git a/js/components/src/components/ui/select.tsx b/js/components/src/components/ui/select.tsx index 890f35bd..4cb592a7 100644 --- a/js/components/src/components/ui/select.tsx +++ b/js/components/src/components/ui/select.tsx @@ -1,9 +1,8 @@ import { Check, ChevronDown } from "lucide-react-native"; import { forwardRef } from "react"; import { View } from "react-native"; -import { zero } from "../.."; import { useTheme } from "../../lib/theme/theme"; -import { flex } from "../../ui"; +import { flex, gap, py } from "../../ui"; import { DropdownMenu, DropdownMenuItem, @@ -13,8 +12,6 @@ import { } from "./dropdown"; import { Text } from "./text"; -const { layout, px, py, borders, r, gap } = zero; - export interface SelectItem { label: string; value: string; diff --git a/js/docs/src/content/docs/lex-reference/media/place-stream-media-getuploadstatus.md b/js/docs/src/content/docs/lex-reference/media/place-stream-media-getuploadstatus.md new file mode 100644 index 00000000..437a9e04 --- /dev/null +++ b/js/docs/src/content/docs/lex-reference/media/place-stream-media-getuploadstatus.md @@ -0,0 +1,131 @@ +--- +title: place.stream.media.getUploadStatus +description: Reference for the place.stream.media.getUploadStatus lexicon +--- + +**Lexicon Version:** 1 + +## Definitions + + + +### `main` + +**Type:** `query` + +Get the processing status of a previously created upload. Only accessible by the DID that created the upload. + +**Parameters:** + +| Name | Type | Req'd | Description | Constraints | +| ---------- | -------- | ----- | ---------------------------------------------------------- | ----------- | +| `uploadId` | `string` | ✅ | The upload ID returned by place.stream.media.createUpload. | | + +**Output:** + +- **Encoding:** `application/json` +- **Schema:** + +**Schema Type:** `object` + +| Name | Type | Req'd | Description | Constraints | +| ------------ | --------------------------------- | ----- | ------------------------------------------------------------------------------- | ------------------------------------------------------ | +| `status` | `string` | ✅ | Current processing status of the upload. | Known Values: `pending`, `processing`, `done`, `error` | +| `tracks` | Array of [`#trackRef`](#trackref) | ❌ | Published track records. Present when status is 'done'. | | +| `durationMs` | `integer` | ❌ | Duration of the processed video in milliseconds. Present when status is 'done'. | | +| `error` | `string` | ❌ | Error message. Present when status is 'error'. | | + +**Possible Errors:** + +- `NotFound`: No upload exists with the given ID for the authenticated user. + +--- + + + +### `trackRef` + +**Type:** `object` + +**Properties:** + +| Name | Type | Req'd | Description | Constraints | +| ----- | -------- | ----- | ----------- | ---------------- | +| `uri` | `string` | ✅ | | Format: `at-uri` | +| `cid` | `string` | ✅ | | | + +--- + +## Lexicon Source + +```json +{ + "lexicon": 1, + "id": "place.stream.media.getUploadStatus", + "defs": { + "main": { + "type": "query", + "description": "Get the processing status of a previously created upload. Only accessible by the DID that created the upload.", + "parameters": { + "type": "params", + "required": ["uploadId"], + "properties": { + "uploadId": { + "type": "string", + "description": "The upload ID returned by place.stream.media.createUpload." + } + } + }, + "output": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": ["status"], + "properties": { + "status": { + "type": "string", + "knownValues": ["pending", "processing", "done", "error"], + "description": "Current processing status of the upload." + }, + "tracks": { + "type": "array", + "description": "Published track records. Present when status is 'done'.", + "items": { + "type": "ref", + "ref": "#trackRef" + } + }, + "durationMs": { + "type": "integer", + "description": "Duration of the processed video in milliseconds. Present when status is 'done'." + }, + "error": { + "type": "string", + "description": "Error message. Present when status is 'error'." + } + } + } + }, + "errors": [ + { + "name": "NotFound", + "description": "No upload exists with the given ID for the authenticated user." + } + ] + }, + "trackRef": { + "type": "object", + "required": ["uri", "cid"], + "properties": { + "uri": { + "type": "string", + "format": "at-uri" + }, + "cid": { + "type": "string" + } + } + } + } +} +``` diff --git a/js/docs/src/content/docs/lex-reference/openapi.json b/js/docs/src/content/docs/lex-reference/openapi.json index a9a97dec..3788f91f 100644 --- a/js/docs/src/content/docs/lex-reference/openapi.json +++ b/js/docs/src/content/docs/lex-reference/openapi.json @@ -1820,6 +1820,83 @@ } } }, + "/xrpc/place.stream.media.getUploadStatus": { + "get": { + "summary": "Get the processing status of a previously created upload. Only accessible by the DID that created the upload.", + "operationId": "place.stream.media.getUploadStatus", + "tags": ["place.stream.media"], + "responses": { + "200": { + "description": "Success", + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "status": { + "type": "string", + "description": "Current processing status of the upload." + }, + "tracks": { + "type": "array", + "description": "Published track records. Present when status is 'done'.", + "items": { + "$ref": "#/components/schemas/place.stream.media.getUploadStatus_trackRef" + } + }, + "durationMs": { + "type": "integer", + "description": "Duration of the processed video in milliseconds. Present when status is 'done'." + }, + "error": { + "type": "string", + "description": "Error message. Present when status is 'error'." + } + }, + "required": ["status"] + } + } + } + }, + "400": { + "description": "Bad Request", + "content": { + "application/json": { + "schema": { + "type": "object", + "required": ["error", "message"], + "properties": { + "error": { + "type": "string", + "oneOf": [ + { + "const": "NotFound" + } + ] + }, + "message": { + "type": "string" + } + } + } + } + } + } + }, + "parameters": [ + { + "name": "uploadId", + "in": "query", + "required": true, + "description": "The upload ID returned by place.stream.media.createUpload.", + "schema": { + "type": "string", + "description": "The upload ID returned by place.stream.media.createUpload." + } + } + ] + } + }, "/xrpc/place.stream.media.getVideo": { "get": { "summary": "Get a hydrated view of a place.stream.video record — the record itself plus author info plus aggregated view counts summed across every reporting node we've indexed. View counts come from place.stream.media.viewCount records; consumers see one number per metric, with the underlying multi-reporter detail collapsed away.", @@ -4277,6 +4354,19 @@ }, "required": ["message", "status", "createdAt"] }, + "place.stream.media.getUploadStatus_trackRef": { + "type": "object", + "properties": { + "uri": { + "type": "string", + "format": "uri" + }, + "cid": { + "type": "string" + } + }, + "required": ["uri", "cid"] + }, "place.stream.media.getVideo_videoView": { "type": "object", "properties": { diff --git a/lexicons/place/stream/media/getUploadStatus.json b/lexicons/place/stream/media/getUploadStatus.json new file mode 100644 index 00000000..3021f7d4 --- /dev/null +++ b/lexicons/place/stream/media/getUploadStatus.json @@ -0,0 +1,75 @@ +{ + "lexicon": 1, + "id": "place.stream.media.getUploadStatus", + "defs": { + "main": { + "type": "query", + "description": "Get the processing status of a previously created upload. Only accessible by the DID that created the upload.", + "parameters": { + "type": "params", + "required": ["uploadId"], + "properties": { + "uploadId": { + "type": "string", + "description": "The upload ID returned by place.stream.media.createUpload." + } + } + }, + "output": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": ["status"], + "properties": { + "status": { + "type": "string", + "knownValues": ["pending", "processing", "done", "error"], + "description": "Current processing status of the upload." + }, + "progress": { + "type": "integer", + "minimum": 0, + "maximum": 100, + "description": "Processing progress percentage (0-100). Only meaningful when status is 'processing'." + }, + "tracks": { + "type": "array", + "description": "Published track records. Present when status is 'done'.", + "items": { + "type": "ref", + "ref": "#trackRef" + } + }, + "durationMs": { + "type": "integer", + "description": "Duration of the processed video in milliseconds. Present when status is 'done'." + }, + "error": { + "type": "string", + "description": "Error message. Present when status is 'error'." + } + } + } + }, + "errors": [ + { + "name": "NotFound", + "description": "No upload exists with the given ID for the authenticated user." + } + ] + }, + "trackRef": { + "type": "object", + "required": ["uri", "cid"], + "properties": { + "uri": { + "type": "string", + "format": "at-uri" + }, + "cid": { + "type": "string" + } + } + } + } +} diff --git a/lexicons/place/stream/media/getVideoList.json b/lexicons/place/stream/media/getVideoList.json new file mode 100644 index 00000000..b1f08faf --- /dev/null +++ b/lexicons/place/stream/media/getVideoList.json @@ -0,0 +1,58 @@ +{ + "lexicon": 1, + "id": "place.stream.media.getVideoList", + "defs": { + "main": { + "type": "query", + "description": "List videos for a given repo DID, newest first. Returns hydrated video views with author info and view counts.", + "parameters": { + "type": "params", + "required": ["repo"], + "properties": { + "repo": { + "type": "string", + "format": "did", + "description": "DID of the repo whose videos to list." + }, + "limit": { + "type": "integer", + "minimum": 1, + "maximum": 100, + "default": 25, + "description": "Maximum number of videos to return." + }, + "cursor": { + "type": "string", + "description": "Pagination cursor from a previous response." + } + } + }, + "output": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": ["videos"], + "properties": { + "videos": { + "type": "array", + "items": { + "type": "ref", + "ref": "lex:place.stream.media.getVideo#videoView" + } + }, + "cursor": { + "type": "string", + "description": "Pagination cursor for the next page, if any." + } + } + } + }, + "errors": [ + { + "name": "RepoNotFound", + "description": "No repo indexed at the supplied DID." + } + ] + } + } +} diff --git a/pkg/model/model.go b/pkg/model/model.go index 51c242f3..98a6fa35 100644 --- a/pkg/model/model.go +++ b/pkg/model/model.go @@ -156,6 +156,7 @@ type Model interface { DeleteMediaViewCount(ctx context.Context, uri string) error GetMediaViewCountByURI(ctx context.Context, uri string) (*streamplace.MediaViewCount, error) GetVideoView(ctx context.Context, uri string) (*streamplace.MediaGetVideo_VideoView, error) + GetVideoList(ctx context.Context, repoDID string, limit int, cursor string) (*streamplace.MediaGetVideoList_Output, error) } var DBRevision = 4 diff --git a/pkg/model/video_list.go b/pkg/model/video_list.go new file mode 100644 index 00000000..875fe6b5 --- /dev/null +++ b/pkg/model/video_list.go @@ -0,0 +1,101 @@ +package model + +import ( + "context" + "fmt" + + "github.com/bluesky-social/indigo/api/bsky" + lexutil "github.com/bluesky-social/indigo/lex/util" + "gorm.io/gorm" + "stream.place/streamplace/pkg/streamplace" +) + +// GetVideoList returns a page of hydrated video views for the given +// repo DID, newest first. Pagination is cursor-based: each page +// returns a cursor for the next page if more videos exist. +func (m *DBModel) GetVideoList(ctx context.Context, repoDID string, limit int, cursor string) (*streamplace.MediaGetVideoList_Output, error) { + if limit <= 0 || limit > 100 { + limit = 25 + } + + query := m.DB.WithContext(ctx). + Model(&Video{}). + Where("repo_did = ?", repoDID). + Order("indexed_at DESC, uri DESC") + + if cursor != "" { + // Cursor is the uri of the last video from the previous page. + // Find its indexed_at to use as the pagination anchor. + var last Video + if err := m.DB.WithContext(ctx). + Where("uri = ?", cursor). + First(&last).Error; err != nil { + return nil, fmt.Errorf("resolve cursor: %w", err) + } + query = query.Where( + "indexed_at < ? OR (indexed_at = ? AND uri < ?)", + last.IndexedAt, last.IndexedAt, cursor, + ) + } + + var rows []*Video + if err := query.Limit(limit + 1).Find(&rows).Error; err != nil { + if err == gorm.ErrRecordNotFound { + return &streamplace.MediaGetVideoList_Output{Videos: []*streamplace.MediaGetVideo_VideoView{}}, nil + } + return nil, fmt.Errorf("list videos: %w", err) + } + + hasMore := len(rows) > limit + if hasMore { + rows = rows[:limit] + } + + // Hydrate each row into a VideoView. + videos := make([]*streamplace.MediaGetVideo_VideoView, 0, len(rows)) + for _, row := range rows { + view, err := m.hydrateVideoView(ctx, row) + if err != nil { + return nil, fmt.Errorf("hydrate video %s: %w", row.URI, err) + } + videos = append(videos, view) + } + + out := &streamplace.MediaGetVideoList_Output{Videos: videos} + if hasMore && len(rows) > 0 { + lastURI := rows[len(rows)-1].URI + out.Cursor = &lastURI + } + return out, nil +} + +// hydrateVideoView builds a VideoView from a model row: decodes the +// record, fetches the author handle, and sums view counts. +func (m *DBModel) hydrateVideoView(ctx context.Context, row *Video) (*streamplace.MediaGetVideo_VideoView, error) { + rec, err := row.ToRecord() + if err != nil { + return nil, err + } + + author := &bsky.ActorDefs_ProfileViewBasic{Did: row.RepoDID} + repo, err := m.GetRepo(row.RepoDID) + if err != nil { + return nil, fmt.Errorf("hydrate author repo: %w", err) + } + if repo != nil { + author.Handle = repo.Handle + } + + summary, err := m.viewCountSummary(ctx, row.URI) + if err != nil { + return nil, err + } + + return &streamplace.MediaGetVideo_VideoView{ + Uri: row.URI, + Cid: row.CID, + Author: author, + Record: &lexutil.LexiconTypeDecoder{Val: rec}, + ViewCounts: summary, + }, nil +} diff --git a/pkg/spxrpc/place_stream_media.go b/pkg/spxrpc/place_stream_media.go index 7f11aa9e..26ad576b 100644 --- a/pkg/spxrpc/place_stream_media.go +++ b/pkg/spxrpc/place_stream_media.go @@ -2,7 +2,9 @@ package spxrpc import ( "context" + "encoding/json" "net/http" + "strings" "github.com/labstack/echo/v4" "github.com/streamplace/oatproxy/pkg/oatproxy" @@ -31,6 +33,9 @@ func (s *Server) handlePlaceStreamMediaCreateUpload(ctx context.Context, body *p if s.uploadManager == nil { return nil, echo.NewHTTPError(http.StatusServiceUnavailable, "upload manager not configured") } + if !strings.HasPrefix(body.MimeType, "video/") { + return nil, echo.NewHTTPError(http.StatusBadRequest, "mimeType must be a video/* type") + } filename := "" if body.Filename != nil { @@ -55,6 +60,63 @@ func (s *Server) handlePlaceStreamMediaCreateUpload(ctx context.Context, body *p }, nil } +func (s *Server) handlePlaceStreamMediaGetUploadStatus(ctx context.Context, uploadId string) (*placestream.MediaGetUploadStatus_Output, error) { + session, _ := oatproxy.GetOAuthSession(ctx) + if session == nil { + return nil, echo.NewHTTPError(http.StatusUnauthorized, "oauth session required") + } + if uploadId == "" { + return nil, echo.NewHTTPError(http.StatusBadRequest, "uploadId is required") + } + upload, err := s.statefulDB.GetUpload(ctx, uploadId) + if err != nil { + return nil, echo.NewHTTPError(http.StatusInternalServerError, err.Error()) + } + if upload == nil || upload.RepoDID != session.DID { + return nil, echo.NewHTTPError(http.StatusNotFound, "upload not found") + } + + out := &placestream.MediaGetUploadStatus_Output{} + + switch upload.ProcessingStatus { + case "done": + out.Status = "done" + if upload.DurationMS > 0 { + d := upload.DurationMS + out.DurationMs = &d + } + if upload.TrackURIs != "" { + var refs []struct { + URI string `json:"uri"` + CID string `json:"cid"` + } + if err := json.Unmarshal([]byte(upload.TrackURIs), &refs); err == nil { + for _, r := range refs { + out.Tracks = append(out.Tracks, &placestream.MediaGetUploadStatus_TrackRef{ + Uri: r.URI, + Cid: r.CID, + }) + } + } + } + case "error": + out.Status = "error" + if upload.ProcessingError != "" { + msg := upload.ProcessingError + out.Error = &msg + } + case "processing": + out.Status = "processing" + p := upload.ProcessingProgress + out.Progress = &p + default: + // "" or any other value: upload not yet fully received + out.Status = "pending" + } + + return out, nil +} + // requestBaseURL returns the scheme+host of the inbound HTTP request, used // to construct user-facing URLs that the same client can reach. func (s *Server) requestBaseURL(ctx context.Context) (string, error) { diff --git a/pkg/spxrpc/place_stream_media_getvideolist.go b/pkg/spxrpc/place_stream_media_getvideolist.go new file mode 100644 index 00000000..cccb8e60 --- /dev/null +++ b/pkg/spxrpc/place_stream_media_getvideolist.go @@ -0,0 +1,38 @@ +package spxrpc + +import ( + "context" + "net/http" + + "github.com/bluesky-social/indigo/atproto/syntax" + "github.com/labstack/echo/v4" + + placestream "stream.place/streamplace/pkg/streamplace" +) + +// handlePlaceStreamMediaGetVideoList returns a paginated, hydrated +// list of video records for a given repo DID. +func (s *Server) handlePlaceStreamMediaGetVideoList(ctx context.Context, repo string, limit *int, cursor *string) (*placestream.MediaGetVideoList_Output, error) { + if repo == "" { + return nil, echo.NewHTTPError(http.StatusBadRequest, "repo is required") + } + if _, err := syntax.ParseDID(repo); err != nil { + return nil, echo.NewHTTPError(http.StatusBadRequest, "repo must be a valid DID: "+err.Error()) + } + + l := 25 + if limit != nil && *limit > 0 && *limit <= 100 { + l = *limit + } + + c := "" + if cursor != nil { + c = *cursor + } + + out, err := s.model.GetVideoList(ctx, repo, l, c) + if err != nil { + return nil, echo.NewHTTPError(http.StatusInternalServerError, err.Error()) + } + return out, nil +} diff --git a/pkg/spxrpc/stubs.go b/pkg/spxrpc/stubs.go index 0dedc81e..4fcc8163 100644 --- a/pkg/spxrpc/stubs.go +++ b/pkg/spxrpc/stubs.go @@ -8,6 +8,7 @@ import ( appbsky "github.com/bluesky-social/indigo/api/bsky" "github.com/labstack/echo/v4" "go.opentelemetry.io/otel" + "net/http" placestream "stream.place/streamplace/pkg/streamplace" ) @@ -303,7 +304,9 @@ func (s *Server) RegisterHandlersPlaceStream(e *echo.Echo) error { e.POST("/xrpc/place.stream.live.startLivestream", s.HandlePlaceStreamLiveStartLivestream) e.POST("/xrpc/place.stream.live.stopLivestream", s.HandlePlaceStreamLiveStopLivestream) e.POST("/xrpc/place.stream.media.createUpload", s.HandlePlaceStreamMediaCreateUpload) + e.GET("/xrpc/place.stream.media.getUploadStatus", s.HandlePlaceStreamMediaGetUploadStatus) e.GET("/xrpc/place.stream.media.getVideo", s.HandlePlaceStreamMediaGetVideo) + e.GET("/xrpc/place.stream.media.getVideoList", s.HandlePlaceStreamMediaGetVideoList) e.POST("/xrpc/place.stream.moderation.createBlock", s.HandlePlaceStreamModerationCreateBlock) e.POST("/xrpc/place.stream.moderation.createGate", s.HandlePlaceStreamModerationCreateGate) e.POST("/xrpc/place.stream.moderation.createPin", s.HandlePlaceStreamModerationCreatePin) @@ -694,6 +697,20 @@ func (s *Server) HandlePlaceStreamMediaCreateUpload(c echo.Context) error { return c.JSON(200, out) } +func (s *Server) HandlePlaceStreamMediaGetUploadStatus(c echo.Context) error { + ctx, span := otel.Tracer("server").Start(c.Request().Context(), "HandlePlaceStreamMediaGetUploadStatus") + defer span.End() + uploadId := c.QueryParam("uploadId") + var out *placestream.MediaGetUploadStatus_Output + var handleErr error + // func (s *Server) handlePlaceStreamMediaGetUploadStatus(ctx context.Context,uploadId string) (*placestream.MediaGetUploadStatus_Output, error) + out, handleErr = s.handlePlaceStreamMediaGetUploadStatus(ctx, uploadId) + if handleErr != nil { + return handleErr + } + return c.JSON(200, out) +} + func (s *Server) HandlePlaceStreamMediaGetVideo(c echo.Context) error { ctx, span := otel.Tracer("server").Start(c.Request().Context(), "HandlePlaceStreamMediaGetVideo") defer span.End() @@ -708,6 +725,37 @@ func (s *Server) HandlePlaceStreamMediaGetVideo(c echo.Context) error { return c.JSON(200, out) } +func (s *Server) HandlePlaceStreamMediaGetVideoList(c echo.Context) error { + ctx, span := otel.Tracer("server").Start(c.Request().Context(), "HandlePlaceStreamMediaGetVideoList") + defer span.End() + repo := c.QueryParam("repo") + limitStr := c.QueryParam("limit") + cursorStr := c.QueryParam("cursor") + + var limit *int + if limitStr != "" { + l, err := strconv.Atoi(limitStr) + if err != nil { + return echo.NewHTTPError(http.StatusBadRequest, "limit must be an integer") + } + limit = &l + } + + var cursor *string + if cursorStr != "" { + cursor = &cursorStr + } + + var out *placestream.MediaGetVideoList_Output + var handleErr error + // func (s *Server) handlePlaceStreamMediaGetVideoList(ctx context.Context,repo string, limit *int, cursor *string) (*placestream.MediaGetVideoList_Output, error) + out, handleErr = s.handlePlaceStreamMediaGetVideoList(ctx, repo, limit, cursor) + if handleErr != nil { + return handleErr + } + return c.JSON(200, out) +} + func (s *Server) HandlePlaceStreamModerationCreateBlock(c echo.Context) error { ctx, span := otel.Tracer("server").Start(c.Request().Context(), "HandlePlaceStreamModerationCreateBlock") defer span.End() diff --git a/pkg/statedb/queue_processor.go b/pkg/statedb/queue_processor.go index fce8b6a9..26e17f74 100644 --- a/pkg/statedb/queue_processor.go +++ b/pkg/statedb/queue_processor.go @@ -126,8 +126,17 @@ func (state *StatefulDB) processVODProcessTask(ctx context.Context, task *AppTas "uploadId", t.UploadID, "did", t.RepoDID) return state.CompleteTask(ctx, task.ID) } + if err := state.SetUploadProcessing(ctx, t.UploadID); err != nil { + log.Warn(ctx, "failed to mark upload as processing", "uploadId", t.UploadID, "error", err) + } cid, err := state.vodProcessor(ctx, t) if err != nil { + if ferr := state.SetUploadFailed(ctx, t.UploadID, err.Error()); ferr != nil { + log.Warn(ctx, "failed to mark upload as failed", "uploadId", t.UploadID, "error", ferr) + } + // Complete the task so it doesn't retry — most VOD failures are + // permanent (unsupported codec, corrupted file, etc.). + _ = state.CompleteTask(ctx, task.ID) return fmt.Errorf("vod processing: %w", err) } log.Log(ctx, "vod processed", "uploadId", t.UploadID, "cid", cid) diff --git a/pkg/statedb/upload.go b/pkg/statedb/upload.go index 52ed2ea3..9f2a5e1f 100644 --- a/pkg/statedb/upload.go +++ b/pkg/statedb/upload.go @@ -25,6 +25,17 @@ type Upload struct { CompletedAt *time.Time `gorm:"column:completed_at"` CreatedAt time.Time `gorm:"column:created_at"` UpdatedAt time.Time `gorm:"column:updated_at"` + + // Processing fields — set by the VOD pipeline after the TUS upload finishes. + // ProcessingStatus is "", "processing", "done", or "error". + ProcessingStatus string `gorm:"column:processing_status"` + ProcessingError string `gorm:"column:processing_error"` + ProcessingProgress int `gorm:"column:processing_progress;default:0"` + // TrackURIs is a JSON array of {"uri":"at://...","cid":"..."} objects + // populated once the track records are published and the video is ready + // for the client to create a place.stream.video record. + TrackURIs string `gorm:"column:track_uris"` + DurationMS int64 `gorm:"column:duration_ms"` } func (Upload) TableName() string { @@ -57,6 +68,41 @@ func (state *StatefulDB) CompleteUpload(ctx context.Context, id string, location }).Error } +func (state *StatefulDB) SetUploadProcessing(ctx context.Context, id string) error { + return state.DB.WithContext(ctx).Model(&Upload{}). + Where("id = ?", id). + Updates(map[string]any{ + "processing_status": "processing", + "processing_progress": 0, + }).Error +} + +func (state *StatefulDB) SetUploadProgress(ctx context.Context, id string, progress int) error { + return state.DB.WithContext(ctx).Model(&Upload{}). + Where("id = ?", id). + Update("processing_progress", progress).Error +} + +func (state *StatefulDB) SetUploadProcessed(ctx context.Context, id string, trackURIsJSON string, durationMS int64) error { + return state.DB.WithContext(ctx).Model(&Upload{}). + Where("id = ?", id). + Updates(map[string]any{ + "processing_status": "done", + "processing_progress": 100, + "track_uris": trackURIsJSON, + "duration_ms": durationMS, + }).Error +} + +func (state *StatefulDB) SetUploadFailed(ctx context.Context, id string, errMsg string) error { + return state.DB.WithContext(ctx).Model(&Upload{}). + Where("id = ?", id). + Updates(map[string]any{ + "processing_status": "error", + "processing_error": errMsg, + }).Error +} + // uploadAuthKeySize is the size in bytes of the HMAC key used to sign upload // bearer tokens. 32 bytes is the recommended size for HS256. const uploadAuthKeySize = 32 diff --git a/pkg/streamplace/mediagetUploadStatus.go b/pkg/streamplace/mediagetUploadStatus.go new file mode 100644 index 00000000..b642452b --- /dev/null +++ b/pkg/streamplace/mediagetUploadStatus.go @@ -0,0 +1,46 @@ +// Code generated by cmd/lexgen (see Makefile's lexgen); DO NOT EDIT. + +// Lexicon schema: place.stream.media.getUploadStatus + +package streamplace + +import ( + "context" + + lexutil "github.com/bluesky-social/indigo/lex/util" +) + +// MediaGetUploadStatus_Output is the output of a place.stream.media.getUploadStatus call. +type MediaGetUploadStatus_Output struct { + // durationMs: Duration of the processed video in milliseconds. Present when status is 'done'. + DurationMs *int64 `json:"durationMs,omitempty" cborgen:"durationMs,omitempty"` + // error: Error message. Present when status is 'error'. + Error *string `json:"error,omitempty" cborgen:"error,omitempty"` + // progress: Processing progress percentage (0-100). Only meaningful when status is 'processing'. + Progress *int `json:"progress,omitempty" cborgen:"progress,omitempty"` + // status: Current processing status of the upload. + Status string `json:"status" cborgen:"status"` + // tracks: Published track records. Present when status is 'done'. + Tracks []*MediaGetUploadStatus_TrackRef `json:"tracks,omitempty" cborgen:"tracks,omitempty"` +} + +// MediaGetUploadStatus_TrackRef is a "trackRef" in the place.stream.media.getUploadStatus schema. +type MediaGetUploadStatus_TrackRef struct { + Cid string `json:"cid" cborgen:"cid"` + Uri string `json:"uri" cborgen:"uri"` +} + +// MediaGetUploadStatus calls the XRPC method "place.stream.media.getUploadStatus". +// +// uploadId: The upload ID returned by place.stream.media.createUpload. +func MediaGetUploadStatus(ctx context.Context, c lexutil.LexClient, uploadId string) (*MediaGetUploadStatus_Output, error) { + var out MediaGetUploadStatus_Output + + params := map[string]interface{}{} + params["uploadId"] = uploadId + if err := c.LexDo(ctx, lexutil.Query, "", "place.stream.media.getUploadStatus", params, nil, &out); err != nil { + return nil, err + } + + return &out, nil +} diff --git a/pkg/streamplace/mediagetVideoList.go b/pkg/streamplace/mediagetVideoList.go new file mode 100644 index 00000000..e58f9027 --- /dev/null +++ b/pkg/streamplace/mediagetVideoList.go @@ -0,0 +1,38 @@ +// Code generated by cmd/lexgen (see Makefile's lexgen); DO NOT EDIT. + +// Lexicon schema: place.stream.media.getVideoList + +package streamplace + +import ( + "context" + + lexutil "github.com/bluesky-social/indigo/lex/util" +) + +// MediaGetVideoList_Output is the output of a place.stream.media.getVideoList call. +type MediaGetVideoList_Output struct { + Cursor *string `json:"cursor,omitempty" cborgen:"cursor,omitempty"` + Videos []*MediaGetVideo_VideoView `json:"videos" cborgen:"videos"` +} + +// MediaGetVideoList calls the XRPC method "place.stream.media.getVideoList". +// +// repo: DID of the repo whose videos to list. +func MediaGetVideoList(ctx context.Context, c lexutil.LexClient, repo string, limit *int, cursor *string) (*MediaGetVideoList_Output, error) { + var out MediaGetVideoList_Output + + params := map[string]interface{}{} + params["repo"] = repo + if limit != nil { + params["limit"] = *limit + } + if cursor != nil { + params["cursor"] = *cursor + } + if err := c.LexDo(ctx, lexutil.Query, "", "place.stream.media.getVideoList", params, nil, &out); err != nil { + return nil, err + } + + return &out, nil +} diff --git a/pkg/vod/process.go b/pkg/vod/process.go index 81b02884..2a046e96 100644 --- a/pkg/vod/process.go +++ b/pkg/vod/process.go @@ -18,6 +18,7 @@ import ( "errors" "fmt" "io" + "sync/atomic" "time" "go.opentelemetry.io/otel" @@ -139,6 +140,13 @@ func ProcessVOD(ctx context.Context, cli *config.CLI, state *statedb.StatefulDB, counter := &countingWriter{} final := io.MultiWriter(hasher, counter, staging) + // Start a progress reporter that polls the byte counter and writes + // percentage updates to the DB every 2 seconds. It stops when the + // pipeline completes (progressDone closed) or the context is cancelled. + progressDone := make(chan struct{}) + defer close(progressDone) + go reportProgress(ctx, state, in.UploadID, counter, size, progressDone, 2*time.Second) + // Ephemeral per-upload signing key. Generated here as muxing starts, // used to C2PA-sign every segment, and dropped when this function // returns — so once the upload is processed nobody can mint more @@ -154,11 +162,11 @@ func ProcessVOD(ctx context.Context, cli *config.CLI, state *statedb.StatefulDB, probe, err := streamThroughMuxl(ctx, src, size, final, metaBuilder, signer.SignerInput) if err != nil { recordErr(span, stagePipeline, err) - // staging is aborted by the deferred Close return "", err } + _ = state.SetUploadProgress(ctx, in.UploadID, 85) span.SetAttributes( - attribute.Int64("output_size_bytes", counter.n), + attribute.Int64("output_size_bytes", counter.load()), attribute.Int64("probe_duration_ms", probe.DurationMS), ) if probe.Video != nil { @@ -175,9 +183,9 @@ func ProcessVOD(ctx context.Context, cli *config.CLI, state *statedb.StatefulDB, attribute.Int("probe_audio_channels", probe.Audio.Channels), ) } - spmetrics.VODOutputBytes.Observe(float64(counter.n)) + spmetrics.VODOutputBytes.Observe(float64(counter.load())) - if counter.n == 0 { + if counter.load() == 0 { err := errors.New("muxl produced zero bytes") recordErr(span, stageEmptyOutput, err) return "", err @@ -187,6 +195,7 @@ func ProcessVOD(ctx context.Context, cli *config.CLI, state *statedb.StatefulDB, recordErr(span, stageStagingComplete, err) return "", fmt.Errorf("complete staging upload: %w", err) } + _ = state.SetUploadProgress(ctx, in.UploadID, 90) finalCID := hasher.CID() contentKey := BlobsPrefix + finalCID + ".mp4" @@ -200,32 +209,23 @@ func ProcessVOD(ctx context.Context, cli *config.CLI, state *statedb.StatefulDB, recordErr(span, stageContentAddressCopy, err) return "", fmt.Errorf("finalize: %w", err) } + _ = state.SetUploadProgress(ctx, in.UploadID, 95) - metafile := metaBuilder.Finalize(finalCID, counter.n) + metafile := metaBuilder.Finalize(finalCID, counter.load()) if err := writeMetafile(ctx, store, finalCID, metafile); err != nil { recordErr(span, stageMetafile, err) return "", fmt.Errorf("write metafile: %w", err) } - // A thumbnail is nice-to-have, not load-bearing: a failure here - // (codec quirk, odd segment) shouldn't sink an otherwise-good - // upload, so log and publish without it. - thumbnail, err := generateThumbnail(ctx, store, finalCID, metafile) - if err != nil { - log.Warn(ctx, "VOD thumbnail generation failed; publishing without thumbnail", "error", err) - } - span.SetAttributes(attribute.Bool("thumbnail_generated", len(thumbnail) > 0)) - if err := publishRecords(ctx, publishParams{ cli: cli, state: state, in: in, cid: finalCID, - size: counter.n, + size: counter.load(), mimeType: "video/mp4", probe: probe, signingKey: signer.DIDKey, - thumbnail: thumbnail, }); err != nil { recordErr(span, stagePublish, err) return "", fmt.Errorf("publish records: %w", err) @@ -237,7 +237,7 @@ func ProcessVOD(ctx context.Context, cli *config.CLI, state *statedb.StatefulDB, "cid", finalCID, "url", store.URL(contentKey), "input_size", size, - "output_size", counter.n, + "output_size", counter.load(), "duration_ms", time.Since(startTime).Milliseconds(), ) return finalCID, nil @@ -281,15 +281,44 @@ func finalizeMove(ctx context.Context, store blob.Store, stagingKey, contentKey return nil } -// countingWriter is an io.Writer that tallies bytes written. Used to -// observe output size without buffering or hashing it twice. -type countingWriter struct{ n int64 } +// countingWriter is an io.Writer that tallies bytes written via an +// atomic counter so a progress goroutine can read it concurrently. +type countingWriter struct{ n atomic.Int64 } func (c *countingWriter) Write(p []byte) (int, error) { - c.n += int64(len(p)) + c.n.Add(int64(len(p))) return len(p), nil } +func (c *countingWriter) load() int64 { return c.n.Load() } + +// reportProgress periodically reads the byte counter and writes a +// percentage to the DB until doneCh closes. The estimate is +// counter.bytes / inputSize, capped at 90 so the remaining stages +// (staging complete, finalize, publish) each have visible increments. +func reportProgress(ctx context.Context, state *statedb.StatefulDB, uploadID string, counter *countingWriter, inputSize int64, doneCh <-chan struct{}, interval time.Duration) { + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-doneCh: + return + case <-ticker.C: + if inputSize <= 0 { + continue + } + pct := int(float64(counter.load()) / float64(inputSize) * 100) + if pct > 90 { + pct = 90 + } + if pct < 5 { + pct = 5 + } + _ = state.SetUploadProgress(ctx, uploadID, pct) + } + } +} + // recordErr is a small helper to attach the standard error attributes // to a span + bump the per-stage counter. Both spans and counters need // to be kept in sync for dashboards to work. diff --git a/pkg/vod/publish.go b/pkg/vod/publish.go index 4b82e93c..d8c53690 100644 --- a/pkg/vod/publish.go +++ b/pkg/vod/publish.go @@ -1,8 +1,8 @@ package vod import ( - "bytes" "context" + "encoding/json" "fmt" comatproto "github.com/bluesky-social/indigo/api/atproto" @@ -22,6 +22,12 @@ import ( "stream.place/streamplace/pkg/streamplace" ) +// trackRefJSON is the shape stored in Upload.TrackURIs. +type trackRefJSON struct { + URI string `json:"uri"` + CID string `json:"cid"` +} + // XRPCClient is the subset of indigo's xrpc.Client we actually call. // Pulled out as an interface so tests / dev wrappers can substitute. // Mirrors pkg/director's XRPCClient for consistency. @@ -29,23 +35,17 @@ type XRPCClient interface { Do(ctx context.Context, method string, contentType string, path string, queryParams map[string]any, body any, out any) error } -// publishParams bundles everything needed to publish the four records -// (one origin + two tracks + one video) that describe a processed VOD. +// publishParams bundles everything needed to publish the origin + track +// records for a processed VOD and store the results on the Upload row. type publishParams struct { - cli *config.CLI - state *statedb.StatefulDB - in Input - cid string - size int64 - mimeType string - probe media.VODResult - // signingKey is the did:key of the ephemeral key that C2PA-signed - // this upload's segments. Recorded on every track record. + cli *config.CLI + state *statedb.StatefulDB + in Input + cid string + size int64 + mimeType string + probe media.VODResult signingKey string - // thumbnail is a JPEG generated from ~halfway through the video, or - // nil if generation failed. Uploaded to the user's PDS and attached - // to the place.stream.video record when present. - thumbnail []byte } // publishRecords does the post-processing record publish: @@ -54,14 +54,12 @@ type publishParams struct { // this blob is fetchable from us). Idempotent: rkey is the CID. // 2. place.stream.media.track in the USER's repo, one per A/V track, // via the user's stored OAuth session. -// 3. place.stream.video in the USER's repo, source = sourceTracks -// referencing the strongRefs returned by step 2. +// 3. Stores the resulting track URIs + duration on the Upload row so +// the client can poll getUploadStatus and create the +// place.stream.video record itself (with full metadata) via Publish. // -// Errors are surfaced to the caller; the calling task processor's -// retry behavior will then re-run the whole pipeline. The origin -// record is idempotent on retry; the track + video records are not -// yet (a retry would produce duplicates). We accept that for V1; a -// follow-up will track created rkeys on the Upload row. +// The video record is intentionally NOT created here — the client +// controls when it becomes visible and supplies the metadata. func publishRecords(ctx context.Context, p publishParams) error { ctx, span := vodTracer.Start(ctx, "vod.publishRecords", trace.WithAttributes( attribute.String("cid", p.cid), @@ -82,7 +80,7 @@ func publishRecords(ctx context.Context, p publishParams) error { return fmt.Errorf("get user xrpc client: %w", err) } - var sourceTracks []*comatproto.RepoStrongRef + var trackRefs []trackRefJSON if p.probe.Video != nil { ref, err := publishTrack(ctx, client, p.in.RepoDID, p.cid, p.size, p.probe.DurationMS, "1", "video", p.signingKey, p.probe.Video, nil) if err != nil { @@ -90,7 +88,7 @@ func publishRecords(ctx context.Context, p publishParams) error { span.SetStatus(codes.Error, "video_track") return fmt.Errorf("publish video track: %w", err) } - sourceTracks = append(sourceTracks, ref) + trackRefs = append(trackRefs, trackRefJSON{URI: ref.Uri, CID: ref.Cid}) } if p.probe.Audio != nil { ref, err := publishTrack(ctx, client, p.in.RepoDID, p.cid, p.size, p.probe.DurationMS, "2", "audio", p.signingKey, nil, p.probe.Audio) @@ -99,15 +97,23 @@ func publishRecords(ctx context.Context, p publishParams) error { span.SetStatus(codes.Error, "audio_track") return fmt.Errorf("publish audio track: %w", err) } - sourceTracks = append(sourceTracks, ref) + trackRefs = append(trackRefs, trackRefJSON{URI: ref.Uri, CID: ref.Cid}) } - if err := publishVideo(ctx, client, p.in, p.probe, sourceTracks, p.thumbnail); err != nil { + trackURIsJSON, err := json.Marshal(trackRefs) + if err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, "marshal_tracks") + return fmt.Errorf("marshal track refs: %w", err) + } + if err := p.state.SetUploadProcessed(ctx, p.in.UploadID, string(trackURIsJSON), p.probe.DurationMS); err != nil { span.RecordError(err) - span.SetStatus(codes.Error, "video") - return fmt.Errorf("publish video: %w", err) + span.SetStatus(codes.Error, "store_tracks") + return fmt.Errorf("store track refs: %w", err) } + span.SetAttributes(attribute.Int("track_count", len(trackRefs))) + span.SetStatus(codes.Ok, "") return nil } @@ -234,71 +240,6 @@ func publishTrack(ctx context.Context, client XRPCClient, did, cid string, blobS }, nil } -// publishVideo creates the top-level place.stream.video record in the -// user's repo, referencing the track records via sourceTracks. The -// title defaults to the upload's filename hint, or "Untitled" if the -// client didn't send one. When thumbnail bytes are supplied they're -// uploaded to the user's PDS and attached as the record's thumb blob; -// an upload failure is logged but doesn't fail the publish. -func publishVideo(ctx context.Context, client XRPCClient, in Input, probe media.VODResult, tracks []*comatproto.RepoStrongRef, thumbnail []byte) error { - ctx, span := vodTracer.Start(ctx, "vod.publishVideo", trace.WithAttributes( - attribute.Int("track_count", len(tracks)), - attribute.Int64("duration_ms", probe.DurationMS), - )) - defer span.End() - - title := in.Filename - if title == "" { - title = "Untitled" - } - duration := probe.DurationMS - - rec := &streamplace.Video{ - LexiconTypeID: constants.PLACE_STREAM_VIDEO, - Title: title, - DurationMs: duration, - Source: &streamplace.Video_Source{ - MediaDefs_SourceTracks: &streamplace.MediaDefs_SourceTracks{ - LexiconTypeID: "place.stream.media.defs#sourceTracks", - Tracks: tracks, - }, - }, - } - - if len(thumbnail) > 0 { - var uploadOut comatproto.RepoUploadBlob_Output - if err := client.Do(ctx, xrpc.Procedure, thumbnailMimeType, "com.atproto.repo.uploadBlob", nil, bytes.NewReader(thumbnail), &uploadOut); err != nil { - log.Warn(ctx, "failed to upload VOD thumbnail blob; publishing without thumbnail", "error", err) - } else { - rec.Thumb = uploadOut.Blob - span.SetAttributes(attribute.Bool("thumb_uploaded", true)) - } - } - - rkey := spid.TIDClock.Next().String() - inp := comatproto.RepoPutRecord_Input{ - Collection: constants.PLACE_STREAM_VIDEO, - Record: &lexutil.LexiconTypeDecoder{Val: rec}, - Rkey: rkey, - Repo: in.RepoDID, - } - out := comatproto.RepoPutRecord_Output{} - if err := client.Do(ctx, xrpc.Procedure, "application/json", "com.atproto.repo.putRecord", map[string]any{}, inp, &out); err != nil { - span.RecordError(err) - return fmt.Errorf("putRecord video: %w", err) - } - span.SetAttributes( - attribute.String("uri", out.Uri), - attribute.String("cid", out.Cid), - ) - log.Log(ctx, "published video record", - "title", title, - "uri", out.Uri, - "duration_ms", duration, - ) - return nil -} - // getUserXRPCClient resolves the user's stored OAuth session, refreshes // it if expiring, and returns an authenticated xrpc client targeting // the user's PDS. Mirrors the pattern used by finalize_livestream and