feat(snapshots): AE scene-snapshot pipeline + admin trigger (Epic C, C1)
Build backend images / build content-svc (push) Failing after 31s
Build backend images / build file-svc (push) Failing after 30s
Build backend images / build gateway (push) Failing after 30s
Build backend images / build identity-svc (push) Failing after 30s
Build backend images / build notification-svc (push) Failing after 31s
Build backend images / build render-svc (push) Failing after 31s
Build backend images / build studio-svc (push) Failing after 31s

Per-scene preview thumbnails for templates. Admin clicks "ساخت پیش‌نمایش
صحنه‌ها" → one single-frame AE render per scene → content.scenes.snapshot_url
→ shown as a thumbnail in the admin scene list (and available to the studio).

- migration 30_render_snapshot_jobs.sql: render.snapshot_jobs (queued|running|
  done|error, per scene, image_url).
- render-svc: db/snapshotjobs.go (EnqueueSceneSnapshots, List, Claim, SetResult
  -> writes content.scenes.snapshot_url cross-schema, SetError); handlers/
  snapshotjobs.go (admin POST/GET /v1/scene-snapshots/:project_id + node-facing
  internal claim/result/fail); main.go routes; gateway route.
- devworker: RunSnapshots — fulfils snapshot jobs with a generated placeholder
  PNG (data: URL, scene-key-tinted) so the flow is verifiable without an AE node.
  Gated by RENDER_DEV_SNAPSHOTS (default off; never hijacks real render jobs).
- admin UI: ProjectScenes "generate snapshots" button (enqueue + poll + reload)
  and a thumbnail (snapshot_url || image) per scene row.

Verified e2e via the dev mock: enqueue -> jobs run -> content.scenes.snapshot_url
populated -> scenes API returns it -> admin renders the thumbnail.

Remaining (C2): node-agent real-AE runner — claim snapshot, aerender -s0 -e0 ->
ffmpeg still -> upload to a PERMANENT URL (mirror file-svc, not the time-limited
export presign) -> post result. Needs a live AE node to build + verify.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
soroush.asadi
2026-06-11 09:54:42 +03:30
parent 93411da462
commit 8488acb115
8 changed files with 406 additions and 0 deletions
+1
View File
@@ -141,6 +141,7 @@ func main() {
v1.Any("/admin-exports/*path", apiRL, auth, render.Handler())
v1.Any("/admin-renders", apiRL, auth, render.Handler())
v1.Any("/template-bundles/*path", apiRL, auth, render.Handler())
v1.Any("/scene-snapshots/*path", apiRL, auth, render.Handler())
v1.Any("/template-scans/*path", apiRL, auth, render.Handler())
v1.Any("/template-scan-jobs/*path", apiRL, auth, render.Handler())
v1.Any("/node-updates/*path", apiRL, auth, render.Handler())
+15
View File
@@ -82,6 +82,7 @@ func main() {
fontH := handlers.NewFontHandler(store)
bundleH := handlers.NewTemplateBundleHandler(mc, minioTemplatesBucket)
scanH := handlers.NewScanHandler(store, mc, minioTemplatesBucket)
snapJobH := handlers.NewSnapshotJobHandler(store, mc, minioTemplatesBucket)
internalH := handlers.NewInternalHandler(store, notifyClient, mc, minioTemplatesBucket, minioBucket)
// ── Dev mock worker (no AE node needed) ────────────────────────────────────
@@ -90,6 +91,11 @@ func main() {
if devWorker {
go devworker.New(store).Run(context.Background())
}
// Snapshot-only dev mock: fulfils scene-snapshot jobs with a generated
// placeholder (no AE), gated separately so it never hijacks real render jobs.
if getEnv("RENDER_DEV_SNAPSHOTS", "false") == "true" {
go devworker.New(store).RunSnapshots(context.Background())
}
// ── Router ────────────────────────────────────────────────────────────────
r := gin.Default()
@@ -174,6 +180,10 @@ func main() {
v1.POST("/template-bundles/:project_id", auth, admin, bundleH.Set)
// ── Template scans (admin: read scenes/colours/configs from the AEP) ───────
// ── Scene snapshots (admin: render one frame per scene from AE) ────────────
v1.POST("/scene-snapshots/:project_id", auth, admin, snapJobH.Enqueue)
v1.GET("/scene-snapshots/:project_id", auth, admin, snapJobH.List)
v1.POST("/template-scans/:project_id/quick", auth, admin, scanH.QuickScan) // headless Go quick-scan
v1.POST("/template-scans/:project_id/jobs", auth, admin, scanH.CreateJob) // queue an AE full scan
v1.GET("/template-scan-jobs/:id", auth, admin, scanH.GetJob)
@@ -205,6 +215,11 @@ func main() {
internal.POST("/render/jobs/:job_id/crash", internalH.Crash)
internal.POST("/render/jobs/:job_id/replica-ready", internalH.ReplicaReady)
// AE scene snapshots (node claims, renders one frame, posts the image URL)
internal.POST("/snapshot/claim", snapJobH.Claim)
internal.POST("/snapshot/:id/result", snapJobH.Result)
internal.POST("/snapshot/:id/fail", snapJobH.Fail)
// AE scan jobs (node claims, runs scan.jsx, posts the ScanResult back)
internal.POST("/scan/claim", scanH.Claim)
internal.GET("/scan/:id/status", scanH.Status) // node watchdog (cancel detection)
+125
View File
@@ -0,0 +1,125 @@
package db
import (
"context"
"github.com/google/uuid"
"github.com/jackc/pgx/v5"
)
// SnapshotJob is an async "render one frame of a scene" job (status row).
type SnapshotJob struct {
ID uuid.UUID `json:"id"`
SceneID uuid.UUID `json:"scene_id"`
SceneKey string `json:"scene_key"`
Status string `json:"status"` // queued | running | done | error
ImageURL *string `json:"image_url,omitempty"`
Error *string `json:"error,omitempty"`
}
// SnapshotClaim is the minimal info a node needs to render a claimed snapshot.
type SnapshotClaim struct {
ID uuid.UUID
ProjectID uuid.UUID
SceneID uuid.UUID
SceneKey string
CompName string
Frame int
}
// EnqueueSceneSnapshots clears any prior snapshot jobs for the project and queues
// one fresh job per active scene (comp_name defaults to the scene key — a comp of
// that name; the node falls back to the render comp when it does not exist).
// Returns the number of jobs queued.
func (s *Store) EnqueueSceneSnapshots(ctx context.Context, projectID uuid.UUID) (int, error) {
tx, err := s.pool.Begin(ctx)
if err != nil {
return 0, err
}
defer tx.Rollback(ctx)
if _, err = tx.Exec(ctx, `DELETE FROM render.snapshot_jobs WHERE project_id = $1`, projectID); err != nil {
return 0, err
}
tag, err := tx.Exec(ctx, `
INSERT INTO render.snapshot_jobs (project_id, scene_id, scene_key, comp_name, frame, status)
SELECT $1, sc.id, sc.key, sc.key, 0, 'queued'
FROM content.scenes sc
WHERE sc.project_id = $1 AND sc.deleted_at IS NULL AND sc.is_active = true`, projectID)
if err != nil {
return 0, err
}
if err = tx.Commit(ctx); err != nil {
return 0, err
}
return int(tag.RowsAffected()), nil
}
// ListSnapshotJobs returns the snapshot jobs for a project (for polling).
func (s *Store) ListSnapshotJobs(ctx context.Context, projectID uuid.UUID) ([]SnapshotJob, error) {
rows, err := s.pool.Query(ctx,
`SELECT id, scene_id, scene_key, status, image_url, error
FROM render.snapshot_jobs WHERE project_id = $1 ORDER BY created_at`, projectID)
if err != nil {
return nil, err
}
defer rows.Close()
var out []SnapshotJob
for rows.Next() {
var j SnapshotJob
if err := rows.Scan(&j.ID, &j.SceneID, &j.SceneKey, &j.Status, &j.ImageURL, &j.Error); err != nil {
return nil, err
}
out = append(out, j)
}
return out, rows.Err()
}
// ClaimSnapshotJob atomically grabs the oldest queued snapshot for a node.
// Returns nil when the queue is empty.
func (s *Store) ClaimSnapshotJob(ctx context.Context, nodeID uuid.UUID) (*SnapshotClaim, error) {
var c SnapshotClaim
err := s.pool.QueryRow(ctx, `
UPDATE render.snapshot_jobs SET status = 'running', node_id = $1, updated_at = NOW()
WHERE id = (
SELECT id FROM render.snapshot_jobs
WHERE status = 'queued'
ORDER BY created_at
LIMIT 1 FOR UPDATE SKIP LOCKED
)
RETURNING id, project_id, scene_id, scene_key, comp_name, frame`,
nodeID).Scan(&c.ID, &c.ProjectID, &c.SceneID, &c.SceneKey, &c.CompName, &c.Frame)
if err != nil {
if err == pgx.ErrNoRows {
return nil, nil
}
return nil, err
}
return &c, nil
}
// SetSnapshotResult marks a running snapshot done with its image URL and writes
// that URL onto the content scene (same DB, cross-schema) so the studio + admin
// render a real thumbnail.
func (s *Store) SetSnapshotResult(ctx context.Context, id uuid.UUID, imageURL string) error {
var sceneID uuid.UUID
err := s.pool.QueryRow(ctx,
`UPDATE render.snapshot_jobs SET status = 'done', image_url = $2, error = NULL, updated_at = NOW()
WHERE id = $1 AND status = 'running' RETURNING scene_id`, id, imageURL).Scan(&sceneID)
if err != nil {
if err == pgx.ErrNoRows {
return nil // job no longer running (cancelled/regenerated) — ignore late result
}
return err
}
_, err = s.pool.Exec(ctx,
`UPDATE content.scenes SET snapshot_url = $2, updated_at = NOW() WHERE id = $1`, sceneID, imageURL)
return err
}
func (s *Store) SetSnapshotError(ctx context.Context, id uuid.UUID, msg string) error {
_, err := s.pool.Exec(ctx,
`UPDATE render.snapshot_jobs SET status = 'error', error = $2, updated_at = NOW() WHERE id = $1 AND status = 'running'`,
id, msg)
return err
}
@@ -102,6 +102,62 @@ func (w *Worker) simulate(ctx context.Context, jobID uuid.UUID) {
log.Printf("[devworker] job %s done", jobID)
}
// ── Snapshot mock ─────────────────────────────────────────────────────────────
// devSnapshotNode is the synthetic node id the mock records on claimed snapshots.
var devSnapshotNode = uuid.MustParse("00000000-0000-0000-0000-0000000000aa")
// RunSnapshots fulfils queued scene-snapshot jobs with a generated placeholder
// image (no AE) so the snapshot flow is exercisable in development. Gated by its
// own flag so it never touches real render jobs. Production uses real nodes.
func (w *Worker) RunSnapshots(ctx context.Context) {
log.Printf("[devworker] snapshot mock started (poll %s) — NOT for production", w.interval)
ticker := time.NewTicker(w.interval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
w.snapTick(ctx)
}
}
}
func (w *Worker) snapTick(ctx context.Context) {
claim, err := w.store.ClaimSnapshotJob(ctx, devSnapshotNode)
if err != nil {
log.Printf("[devworker] snapshot claim error: %v", err)
return
}
if claim == nil {
return // queue empty
}
if err := w.store.SetSnapshotResult(ctx, claim.ID, snapshotPlaceholder(claim.SceneKey)); err != nil {
log.Printf("[devworker] snapshot %s result failed: %v", claim.ID, err)
_ = w.store.SetSnapshotError(ctx, claim.ID, err.Error())
return
}
log.Printf("[devworker] snapshot %s (scene %s) done", claim.ID, claim.SceneKey)
}
// snapshotPlaceholder builds a 480×270 PNG card tinted by the scene key with a
// little "play" block, returned as a data: URL so the dev path needs no storage.
func snapshotPlaceholder(sceneKey string) string {
const w, h = 480, 270
var sum uint32
for _, r := range sceneKey {
sum = sum*31 + uint32(r)
}
base := color.RGBA{uint8(40 + sum%120), uint8(40 + (sum/120)%120), uint8(80 + (sum/7)%150), 255}
img := image.NewRGBA(image.Rect(0, 0, w, h))
draw.Draw(img, img.Bounds(), &image.Uniform{base}, image.Point{}, draw.Src)
draw.Draw(img, image.Rect(0, h/2-30, w, h/2+30), &image.Uniform{color.RGBA{0, 0, 0, 60}}, image.Point{}, draw.Over)
draw.Draw(img, image.Rect(w/2-18, h/2-18, w/2+18, h/2+18), &image.Uniform{color.RGBA{255, 255, 255, 230}}, image.Point{}, draw.Over)
var buf bytes.Buffer
_ = png.Encode(&buf, img)
return "data:image/png;base64," + base64.StdEncoding.EncodeToString(buf.Bytes())
}
// previewB64 builds a 320×180 PNG with a progress bar — same idea as the node
// agent's GeneratePreviewB64, kept local so render-svc has no node-agent dep.
func previewB64(pct int) string {
@@ -0,0 +1,133 @@
package handlers
import (
"net/http"
"github.com/flatrender/render-svc/internal/db"
"github.com/flatrender/render-svc/internal/models"
"github.com/gin-gonic/gin"
"github.com/google/uuid"
"github.com/minio/minio-go/v7"
)
// SnapshotJobHandler queues per-scene single-frame AE snapshot jobs and exposes
// the node-facing claim/result/fail lifecycle. On result it writes the image URL
// onto content.scenes.snapshot_url (in the store, cross-schema).
type SnapshotJobHandler struct {
store *db.Store
minio *minio.Client
templatesBucket string
}
func NewSnapshotJobHandler(store *db.Store, mc *minio.Client, templatesBucket string) *SnapshotJobHandler {
return &SnapshotJobHandler{store: store, minio: mc, templatesBucket: templatesBucket}
}
// POST /v1/scene-snapshots/:project_id (admin) → queue one job per active scene.
func (h *SnapshotJobHandler) Enqueue(c *gin.Context) {
pid, err := uuid.Parse(c.Param("project_id"))
if err != nil {
c.JSON(http.StatusBadRequest, models.APIError{Code: "bad_request", Message: "invalid project_id"})
return
}
n, err := h.store.EnqueueSceneSnapshots(c.Request.Context(), pid)
if err != nil {
c.JSON(http.StatusInternalServerError, models.APIError{Code: "internal_error", Message: err.Error()})
return
}
c.JSON(http.StatusOK, gin.H{"enqueued": n})
}
// GET /v1/scene-snapshots/:project_id (admin) → job statuses for polling.
func (h *SnapshotJobHandler) List(c *gin.Context) {
pid, err := uuid.Parse(c.Param("project_id"))
if err != nil {
c.JSON(http.StatusBadRequest, models.APIError{Code: "bad_request", Message: "invalid project_id"})
return
}
jobs, err := h.store.ListSnapshotJobs(c.Request.Context(), pid)
if err != nil {
c.JSON(http.StatusInternalServerError, models.APIError{Code: "internal_error", Message: err.Error()})
return
}
if jobs == nil {
jobs = []db.SnapshotJob{}
}
c.JSON(http.StatusOK, gin.H{"jobs": jobs})
}
// POST /v1/internal/snapshot/claim (node, HMAC)
func (h *SnapshotJobHandler) Claim(c *gin.Context) {
var req models.ClaimJobRequest
_ = c.ShouldBindJSON(&req)
claim, err := h.store.ClaimSnapshotJob(c.Request.Context(), req.NodeID)
if err != nil {
c.JSON(http.StatusInternalServerError, models.APIError{Code: "internal_error", Message: err.Error()})
return
}
if claim == nil {
c.Status(http.StatusNoContent)
return
}
url, isBundle, md5 := resolveTemplateObject(h.minio, h.templatesBucket, claim.ProjectID)
if url == "" {
_ = h.store.SetSnapshotError(c.Request.Context(), claim.ID,
"no template stored for this project — upload the .aep from «فایل‌ها» first")
c.Status(http.StatusNoContent)
return
}
c.JSON(http.StatusOK, gin.H{
"snapshot_job_id": claim.ID,
"project_id": claim.ProjectID,
"scene_id": claim.SceneID,
"scene_key": claim.SceneKey,
"comp_name": claim.CompName,
"frame": claim.Frame,
"aep_download_url": url,
"is_bundle": isBundle,
"bundle_md5": md5,
})
}
// POST /v1/internal/snapshot/:id/result (node, HMAC) body {image_url}
func (h *SnapshotJobHandler) Result(c *gin.Context) {
id, err := uuid.Parse(c.Param("id"))
if err != nil {
c.JSON(http.StatusBadRequest, models.APIError{Code: "bad_request", Message: "invalid id"})
return
}
var req struct {
ImageURL string `json:"image_url"`
}
if err := c.ShouldBindJSON(&req); err != nil || req.ImageURL == "" {
c.JSON(http.StatusBadRequest, models.APIError{Code: "bad_request", Message: "image_url required"})
return
}
if err := h.store.SetSnapshotResult(c.Request.Context(), id, req.ImageURL); err != nil {
c.JSON(http.StatusInternalServerError, models.APIError{Code: "internal_error", Message: err.Error()})
return
}
c.Status(http.StatusNoContent)
}
// POST /v1/internal/snapshot/:id/fail (node, HMAC) body {reason}
func (h *SnapshotJobHandler) Fail(c *gin.Context) {
id, err := uuid.Parse(c.Param("id"))
if err != nil {
c.JSON(http.StatusBadRequest, models.APIError{Code: "bad_request", Message: "invalid id"})
return
}
var req struct {
Reason string `json:"reason"`
}
_ = c.ShouldBindJSON(&req)
if req.Reason == "" {
req.Reason = "snapshot failed"
}
if err := h.store.SetSnapshotError(c.Request.Context(), id, req.Reason); err != nil {
c.JSON(http.StatusInternalServerError, models.APIError{Code: "internal_error", Message: err.Error()})
return
}
c.Status(http.StatusNoContent)
}