OverviewFilesCommitsIssuesDocsReleasesPeers

hyper-note commits / 2cc42f6

Add VideoProcessor with ffmpeg probe, poster extraction, and processing queue
Spawns ffmpeg/ffprobe CLI via bare-subprocess for video metadata extraction and JPEG poster frame generation. Includes sequential processing queue that writes video index entries to MediaStore's Hyperbee. Safe fps parser replaces eval-based parsing. All 4 tests pass (probe, poster, processBasic, enqueue). Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2cc42f66f907af41031196eab61b3fea8f548b9e
by plankatron <fleeky@gmail.com> · 2026-03-19T18:05:21.000Z

Changed files (2) +376 -0

lib/media/video-processor.js +212 -0
……@@ -0,0 +1,212 @@
1import { spawn } from 'bare-subprocess'
2import fs from 'bare-fs'
3import path from 'bare-path'
4import os from 'bare-os'
5
6// Quality levels for HLS (used later in Task 12)
7export const HLS_LEVELS = [
8 { name: '480p', height: 480, bitrate: '1000k', maxrate: '1200k', profile: 'baseline' },
9 { name: '720p', height: 720, bitrate: '2500k', maxrate: '3000k', profile: 'main' },
10 { name: '1080p', height: 1080, bitrate: '5000k', maxrate: '6000k', profile: 'high' }
11]
12
13export class VideoProcessor {
14 constructor (mediaStore, opts = {}) {
15 this.mediaStore = mediaStore
16 this._queue = []
17 this._processing = false
18 this._tmpBase = opts.tmpDir || path.join(os.tmpdir(), 'hypernote-video')
19 this._ffmpegPath = opts.ffmpegPath || 'ffmpeg'
20 this._ffprobePath = opts.ffprobePath || 'ffprobe'
21 }
22
23 /**
24 * Run a command and return { stdout, stderr, code }
25 */
26 _exec (cmd, args) {
27 return new Promise((resolve, reject) => {
28 const proc = spawn(cmd, args)
29 const stdout = []
30 const stderr = []
31
32 proc.stdout.on('data', (chunk) => stdout.push(chunk))
33 proc.stderr.on('data', (chunk) => stderr.push(chunk))
34
35 proc.on('exit', (code) => {
36 resolve({
37 code,
38 stdout: Buffer.concat(stdout).toString(),
39 stderr: Buffer.concat(stderr).toString()
40 })
41 })
42
43 proc.on('error', reject)
44 })
45 }
46
47 /**
48 * Safely parse ffprobe frame rate string like "30/1" or "25"
49 */
50 _parseFps (str) {
51 if (!str) return 30
52 const parts = str.split('/')
53 const num = Number(parts[0])
54 const den = parts.length > 1 ? Number(parts[1]) : 1
55 if (!num || !den || !isFinite(num / den)) return 30
56 return Math.round(num / den)
57 }
58
59 /**
60 * Probe a video file — returns { width, height, duration, codec, fps }
61 */
62 async probe (filePath) {
63 const result = await this._exec(this._ffprobePath, [
64 '-v', 'quiet',
65 '-print_format', 'json',
66 '-show_streams',
67 '-show_format',
68 filePath
69 ])
70
71 if (result.code !== 0) {
72 throw new Error('ffprobe failed: ' + result.stderr)
73 }
74
75 const data = JSON.parse(result.stdout)
76 const video = data.streams.find(s => s.codec_type === 'video')
77 if (!video) throw new Error('No video stream found')
78
79 const duration = parseFloat(data.format.duration || video.duration || '0')
80 const fps = this._parseFps(video.r_frame_rate)
81
82 return {
83 width: video.width,
84 height: video.height,
85 duration,
86 codec: video.codec_name,
87 fps
88 }
89 }
90
91 /**
92 * Extract a poster frame at the given time (seconds) as JPEG buffer.
93 */
94 async extractPoster (filePath, timeSeconds) {
95 const time = timeSeconds || 2
96 const tmpPoster = path.join(this._tmpBase, 'poster-' + Date.now() + '.jpg')
97
98 await fs.promises.mkdir(this._tmpBase, { recursive: true })
99
100 const result = await this._exec(this._ffmpegPath, [
101 '-y',
102 '-ss', String(time),
103 '-i', filePath,
104 '-vframes', '1',
105 '-q:v', '2',
106 tmpPoster
107 ])
108
109 if (result.code !== 0) {
110 throw new Error('Poster extraction failed: ' + result.stderr)
111 }
112
113 const buf = await fs.promises.readFile(tmpPoster)
114 await fs.promises.unlink(tmpPoster).catch(() => {})
115 return buf
116 }
117
118 /**
119 * Write a MediaStore blob to a temp file (ffmpeg needs file paths).
120 * Returns the temp file path. Caller must clean up.
121 */
122 async _writeTempFile (hash, ext) {
123 const buf = await this.mediaStore.get(hash)
124 if (!buf) throw new Error('Blob not found: ' + hash)
125
126 await fs.promises.mkdir(this._tmpBase, { recursive: true })
127 const tmpPath = path.join(this._tmpBase, hash.substring(0, 16) + (ext || '.mp4'))
128 await fs.promises.writeFile(tmpPath, buf)
129 return tmpPath
130 }
131
132 /**
133 * Process a video: probe it, extract poster, store poster in MediaStore.
134 * Returns { probe, posterHash }
135 */
136 async processBasic (originalHash) {
137 // Write to temp file
138 const info = await this.mediaStore.getInfo(originalHash)
139 const ext = info && info.mimeType === 'video/webm' ? '.webm' : '.mp4'
140 const tmpPath = await this._writeTempFile(originalHash, ext)
141
142 try {
143 // Probe
144 const probeResult = await this.probe(tmpPath)
145
146 // Extract poster
147 const posterTime = Math.min(2, probeResult.duration * 0.1)
148 const posterBuf = await this.extractPoster(tmpPath, posterTime)
149 const posterHash = await this.mediaStore.put(posterBuf, 'image/jpeg')
150
151 return { probe: probeResult, posterHash }
152 } finally {
153 await fs.promises.unlink(tmpPath).catch(() => {})
154 }
155 }
156
157 /**
158 * Enqueue a video for full processing (probe + poster + HLS).
159 * Returns a promise that resolves when processing is complete.
160 */
161 enqueue (originalHash) {
162 return new Promise((resolve, reject) => {
163 this._queue.push({ originalHash, resolve, reject })
164 this._processNext()
165 })
166 }
167
168 async _processNext () {
169 if (this._processing || this._queue.length === 0) return
170 this._processing = true
171
172 const { originalHash, resolve, reject } = this._queue.shift()
173
174 try {
175 const result = await this.processBasic(originalHash)
176
177 // Store video index
178 await this.mediaStore._index.put('video:' + originalHash, {
179 status: 'ready',
180 poster: result.posterHash,
181 width: result.probe.width,
182 height: result.probe.height,
183 duration: result.probe.duration,
184 codec: result.probe.codec,
185 fps: result.probe.fps,
186 processedAt: Date.now()
187 })
188
189 resolve(result)
190 } catch (err) {
191 await this.mediaStore._index.put('video:' + originalHash, {
192 status: 'failed',
193 error: err.message,
194 processedAt: Date.now()
195 }).catch(() => {})
196 reject(err)
197 } finally {
198 this._processing = false
199 this._processNext()
200 }
201 }
202
203 /**
204 * Get video processing status.
205 */
206 async getStatus (originalHash) {
207 const entry = await this.mediaStore._index.get('video:' + originalHash)
208 return entry ? entry.value : null
209 }
210}
211
212export default VideoProcessor
test/unit/video-processor.test.js +164 -0
……@@ -0,0 +1,164 @@
1import test from 'brittle'
2import Corestore from 'corestore'
3import fs from 'bare-fs'
4import path from 'bare-path'
5import os from 'bare-os'
6import { MediaStore } from '../../lib/media-store.js'
7import { VideoProcessor } from '../../lib/media/video-processor.js'
8import { spawn } from 'bare-subprocess'
9
10function tmpdir () {
11 const dir = path.join(os.tmpdir(), 'hn-video-test-' + Date.now() + '-' + Math.random().toString(36).slice(2))
12 fs.mkdirSync(dir, { recursive: true })
13 return dir
14}
15
16function rmrf (dir) {
17 try {
18 for (const f of fs.readdirSync(dir)) {
19 const fp = path.join(dir, f)
20 const stat = fs.statSync(fp)
21 if (stat.isDirectory()) rmrf(fp)
22 else fs.unlinkSync(fp)
23 }
24 fs.rmdirSync(dir)
25 } catch {}
26}
27
28// Generate a tiny test video (2s, 320x240, solid color)
29function generateTestVideo (outputPath) {
30 return new Promise((resolve, reject) => {
31 const proc = spawn('ffmpeg', [
32 '-y', '-f', 'lavfi', '-i', 'color=c=blue:s=320x240:d=2',
33 '-f', 'lavfi', '-i', 'anullsrc=r=44100:cl=stereo',
34 '-t', '2',
35 '-c:v', 'libx264', '-profile:v', 'baseline', '-pix_fmt', 'yuv420p',
36 '-c:a', 'aac', '-b:a', '64k',
37 '-shortest',
38 outputPath
39 ])
40 const stderr = []
41 proc.stderr.on('data', (c) => stderr.push(c))
42 proc.on('exit', (code) => {
43 if (code !== 0) reject(new Error('ffmpeg gen failed: ' + Buffer.concat(stderr).toString()))
44 else resolve()
45 })
46 proc.on('error', reject)
47 })
48}
49
50test('VideoProcessor - probe returns video metadata', async (t) => {
51 const dir = tmpdir()
52 try {
53 const videoPath = path.join(dir, 'test.mp4')
54 await generateTestVideo(videoPath)
55
56 const store = new Corestore(path.join(dir, 'store'))
57 const media = new MediaStore(store)
58 await media.ready()
59
60 const vp = new VideoProcessor(media, { tmpDir: path.join(dir, 'tmp') })
61 const info = await vp.probe(videoPath)
62
63 t.is(info.width, 320)
64 t.is(info.height, 240)
65 t.ok(info.duration >= 1.5 && info.duration <= 2.5, 'duration ~2s')
66 t.is(info.codec, 'h264')
67 t.ok(info.fps > 0, 'fps > 0')
68
69 await store.close()
70 } finally {
71 rmrf(dir)
72 }
73})
74
75test('VideoProcessor - extractPoster returns JPEG buffer', async (t) => {
76 const dir = tmpdir()
77 try {
78 const videoPath = path.join(dir, 'test.mp4')
79 await generateTestVideo(videoPath)
80
81 const store = new Corestore(path.join(dir, 'store'))
82 const media = new MediaStore(store)
83 await media.ready()
84
85 const vp = new VideoProcessor(media, { tmpDir: path.join(dir, 'tmp') })
86 const posterBuf = await vp.extractPoster(videoPath, 1)
87
88 t.ok(posterBuf.length > 0, 'poster buffer not empty')
89 // JPEG magic bytes
90 t.is(posterBuf[0], 0xFF, 'JPEG SOI byte 0')
91 t.is(posterBuf[1], 0xD8, 'JPEG SOI byte 1')
92
93 await store.close()
94 } finally {
95 rmrf(dir)
96 }
97})
98
99test('VideoProcessor - processBasic stores poster and returns probe', async (t) => {
100 const dir = tmpdir()
101 try {
102 const videoPath = path.join(dir, 'test.mp4')
103 await generateTestVideo(videoPath)
104
105 const store = new Corestore(path.join(dir, 'store'))
106 const media = new MediaStore(store)
107 await media.ready()
108
109 // Store the test video in MediaStore
110 const videoBuf = await fs.promises.readFile(videoPath)
111 const hash = await media.put(videoBuf, 'video/mp4')
112
113 const vp = new VideoProcessor(media, { tmpDir: path.join(dir, 'tmp') })
114 const result = await vp.processBasic(hash)
115
116 t.ok(result.probe, 'probe result exists')
117 t.is(result.probe.width, 320)
118 t.is(result.probe.height, 240)
119 t.ok(result.posterHash, 'posterHash exists')
120 t.is(result.posterHash.length, 64, 'posterHash is SHA-256 hex')
121
122 // Verify poster was stored
123 const posterBuf = await media.get(result.posterHash)
124 t.ok(posterBuf, 'poster buffer retrievable')
125 t.is(posterBuf[0], 0xFF, 'poster is JPEG')
126
127 await store.close()
128 } finally {
129 rmrf(dir)
130 }
131})
132
133test('VideoProcessor - enqueue processes and writes video index', async (t) => {
134 const dir = tmpdir()
135 try {
136 const videoPath = path.join(dir, 'test.mp4')
137 await generateTestVideo(videoPath)
138
139 const store = new Corestore(path.join(dir, 'store'))
140 const media = new MediaStore(store)
141 await media.ready()
142
143 const videoBuf = await fs.promises.readFile(videoPath)
144 const hash = await media.put(videoBuf, 'video/mp4')
145
146 const vp = new VideoProcessor(media, { tmpDir: path.join(dir, 'tmp') })
147 const result = await vp.enqueue(hash)
148
149 t.ok(result.probe, 'probe result exists')
150 t.ok(result.posterHash, 'posterHash exists')
151
152 // Check video index was written
153 const status = await vp.getStatus(hash)
154 t.is(status.status, 'ready')
155 t.is(status.width, 320)
156 t.is(status.height, 240)
157 t.ok(status.poster, 'poster hash in index')
158 t.ok(status.duration > 0, 'duration > 0')
159
160 await store.close()
161 } finally {
162 rmrf(dir)
163 }
164})